在文件存储 HDFS 上使用 Apache Flink

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
简介: 本文主要为大家介绍在文件存储HDFS上搭建及使用Apache Flink的方法。

111.jpg
镜像下载、域名解析、时间同步请点击 阿里巴巴开源镜像站

一、准备工作

在文件存储HDFS上使用Apache Flink,需要先完成以下准备工作。

说明 本文档的操作步骤中涉及的安装包版本号、文件夹路径,请根据实际情况进行替换。

  1. 开通文件存储HDFS服务并创建文件系统实例和挂载点,详情请参见HDFS快速入门
  2. 在计算节点上安装JDK。版本不能低于1.8。
  3. 在计算节点上安装Scala。Scala下载地址:官方链接,其版本要与使用的Apache Flink版本相兼容。
  4. 下载Apache Hadoop压缩包。Apache Hadoop下载地址:官方链接。建议您选用的Apache Hadoop版本不低于2.7.2,本文档中使用的Apache Hadoop版本为Apache Hadoop 2.7.2。
  5. 下载Apache Flink压缩包。在文件存储HDFS上使用的Flink的版本必须为1.9.0及以上,Apache Flink下载地址:官方链接。本文档中使用的Flink版本为官方提供的预编译版本Flink 1.9.0。

二、配置Apache Hadoop

1、执行如下命令解压Apache Hadoop压缩包到指定文件夹。

tar -zxvf hadoop-2.7.2.tar.gz -C /usr/local/

2、修改hadoop-env.sh配置文件。

  • 执行如下命令打开hadoop-env.sh配置文件。
vim /usr/local/hadoop-2.7.2/etc/hadoop/hadoop-env.sh
  • 配置JAVA_HOME目录,如下所示。
export JAVA_HOME=/usr/java/default

3、修改core-site.xml配置文件。

  • 执行如下命令打开core-site.xml配置文件。
vim /usr/local/hadoop-2.7.2/etc/hadoop/core-site.xml
  • 在core-site.xml配置文件中,配置如下信息,详情请参见挂载文件系统
<configuration>
<property>
     <name>fs.defaultFS</name>
     <value>dfs://x-xxxxxxxx.cn-xxxxx.dfs.aliyuncs.com:10290</value>
     <!-- 该地址填写您的挂载点地址 -->
</property>
<property>
     <name>fs.dfs.impl</name>
     <value>com.alibaba.dfs.DistributedFileSystem</value>
</property>
<property>
     <name>fs.AbstractFileSystem.dfs.impl</name>
     <value>com.alibaba.dfs.DFS</value>
</property>
<property>
     <name>io.file.buffer.size</name>
     <value>8388608</value>
</property>
<property>
     <name>alidfs.use.buffer.size.setting</name>
     <value>true</value>
</property>
<property>
     <name>dfs.usergroupservice.impl</name>
     <value>com.alibaba.dfs.security.LinuxUserGroupService.class</value>
</property>
  <property>
     <name>dfs.connection.count</name>
     <value>16</value>
</property>
</configuration>

4、修改mapred-site.xml配置文件。

  • 执行如下命令打开mapred-site.xml配置文件。
vim /usr/local/hadoop-2.7.2/etc/hadoop/mapred-site.xml
  • 在mapred-site.xml配置文件中,配置如下信息。
<configuration>
<property>
      <name>mapreduce.framework.name</name>
      <value>yarn</value>
</property>
</configuration>

5、修改yarn-site.xml配置文件。

  • 执行如下命令打开yarn-site.xml配置文件。
vim /usr/local/hadoop-2.7.2/etc/hadoop/yarn-site.xml
  • 在yarn-site.xml配置文件中,配置如下信息。
<configuration>
<property>
  <name>yarn.resourcemanager.hostname</name>
  <value>xxxx</value>
  <!-- 该地址填写集群中yarn的resourcemanager的hostname -->
</property>
<property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle</value>
</property>
<property>
  <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
  <value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>
<property>
  <name>yarn.nodemanager.resource.memory-mb</name>
  <value>16384</value>
    <!-- 根据您当前的集群能力进行配置此项 -->
</property>
<property>
  <name>yarn.nodemanager.resource.cpu-vcores</name>
  <value>4</value>
     <!-- 根据您当前的集群能力进行配置此项 -->
</property>
<property>
  <name>yarn.scheduler.maximum-allocation-vcores</name>
  <value>4</value>
    <!-- 根据您当前的集群能力进行配置此项 -->
</property>
<property>
  <name>yarn.scheduler.minimum-allocation-mb</name>
  <value>3584</value>
    <!-- 根据您当前的集群能力进行配置此项 -->
</property>
<property>
  <name>yarn.scheduler.maximum-allocation-mb</name>
  <value>14336</value>
    <!-- 根据您当前的集群能力进行配置此项 -->
</property>
</configuration>

6、修改slaves配置文件。

  • 执行如下命令打开slaves配置文件。
vim /usr/local/hadoop-2.7.2/etc/hadoop/slaves
  • 在slaves配置文件中,配置如下信息。
node1
node2

7、配置环境变量。

  • 执行如下命令打开/etc/profile配置文件。
vim /etc/profile
  • 在/etc/profile配置文件中,配置如下信息。
export HADOOP_HOME=/usr/local/hadoop-2.7.2
export HADOOP_CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath)
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH
  • 执行如下命令使配置生效。
source /etc/profile

8、执行如下命令配置文件存储HDFS的SDK。您可以单击下载文件存储HDFS的SDK(此处以aliyun-sdk-dfs-1.0.3.jar为例),将其部署在Apache Hadoop生态系统组件的CLASSPATH上,详情请参见挂载文件系统

cp aliyun-sdk-dfs-1.0.3.jar  /usr/local/hadoop-2.7.2/share/hadoop/hdfs

9、执行如下命令将${HADOOP_HOME}文件夹同步到集群的其他节点。

scp -r hadoop-2.7.2/ root@node2:/usr/local/

三、验证Apache Hadoop配置

完成Apache Hadoop配置后,不需要格式化namenode,也不需要使用start-dfs.sh来启动HDFS相关服务。如需使用yarn服务,只需在resourcemanager节点启动yarn服务,具体验证Apache Hadoop配置成功的方法请参见验证安装

四、编译flink-shade

1、下载 flink-shade源码到指定目录。

git clone https://github.com/apache/flink-shaded.git  ~/flink-shade

2、修改flink-shade源码中的pom文件。修改Hadoop版本为您的集群中使用的版本,在本文档中使用的Hadoop版本为2.7.2。

vim  ~/flink-shaded/flink-shaded-hadoop-2-parent/pom.xml

1.png
在依赖项中添加文件存储HDFS SDK,在本文档使用文件存储HDFS SDK版本为1.0.3。

vim  ~/flink-shaded/flink-shaded-hadoop-2-parent/flink-shaded-hadoop-2/pom.xml
...
<dependency>
       <groupId>com.aliyun.dfs</groupId>
       <artifactId>aliyun-sdk-dfs</artifactId>
       <version>1.0.3</version>
</dependency>
...

2.png
3、编译打包。

cd ~/flink-shaded
mvn package -Dshade-sources

五、配置Apache Flink

1、执行如下命令解压Flink压缩包到指定文件夹。

tar -zxvf flink-1.9.0-bin-scala_2.11.tgz -C /usr/local/

2、拷贝flink-shade编译的flink-shaded-hadoop-2-uber-x.y.z.jar到Flink的lib目录下。

cp  ~/flink-shaded/flink-shaded-hadoop-2-parent/flink-shaded-hadoop-2-uber/target/flink-shaded-hadoop-2-uber-2.7.2-11.0.jar /usr/local/flink-1.9.0/lib/

说明

  • 在使用Apache Flink之前必须在您的集群环境变量中配置HADOOP_HOME,HADOOP_CLASSPATH和HADOOP_CONF_DIR,详情请参见配置Apache Hadoop中的步骤7:配置环境变量。
  • 如果您使用的Flink版本中已经包含flink-shaded-hadoop-2-uber-x.y.z.jar,则需要使用编译flink-shade中编译的flink-shaded-hadoop-2-uber-x.y.z.jar进行替换。
  • 如果您需要对Flink进行额外的配置,请参考官方文档:配置操作指南

六、验证Apache Flink配置

使用Flink自带的WordCount.jar对文件存储HDFS上的数据进行读取,并将计算结果写入到文件存储HDFS,在测试之前需要先启动yarn服务。
1、生成测试数据。此处使用Apache Hadoop 2.7.2自带的jar包hadoop-mapreduce-examples-2.7.2.jar中的randomtextwriter方法在文件存储HDFS上生成测试数据。

/usr/local/hadoop-2.7.2/bin/hadoop jar  /usr/local/hadoop-2.7.2/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar 
randomtextwriter \
-D mapreduce.randomtextwriter.totalbytes=10240 \
-D mapreduce.randomtextwriter.bytespermap=1024 \
-D mapreduce.job.maps=4  \
-D mapreduce.job.reduces=2  \
dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290/flink-test/input \

其中,dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290为文件存储HDFS的挂载点,请根据您的实际情况替换。
2、查看在文件存储HDFS上生成的测试数据。

/usr/local/hadoop-2.7.2/bin/hadoop fs -cat dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290/flink-test/input/*

其中,dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290为文件存储HDFS的挂载点,请根据您的实际情况替换。
3、提交wordcount程序。

/usr/local/flink-1.9.0/bin/flink run 
-m yarn-cluster -yn 1 -yjm 1024 -ytm 1024 \
/usr/local/flink-1.9.0/examples/batch/WordCount.jar \
--input dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290/flink-test/input \
--output dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290/flink-test/output \

其中,dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290为文件存储HDFS的挂载点,请根据您的实际情况替换。
4、查看在文件存储HDFS上的结果文件。

/usr/local/hadoop-2.7.2/bin/hadoop fs -cat dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290/flink-test/output

其中,dfs://f-xxxxx.cn-xxx.dfs.aliyuncs.com:10290为文件存储HDFS的挂载点,请根据您的实际情况替换。
3.png

阿里巴巴开源镜像站 提供全面,高效和稳定的镜像下载服务。钉钉搜索 ' 21746399 ‘ 加入镜像站官方用户交流群。”

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
6月前
|
SQL 存储 人工智能
Apache Flink 2.0.0: 实时数据处理的新纪元
Apache Flink 2.0.0 正式发布!这是自 Flink 1.0 发布九年以来的首次重大更新,凝聚了社区两年的努力。此版本引入分离式状态管理、物化表、流批统一等创新功能,优化云原生环境下的资源利用与性能表现,并强化了对人工智能工作流的支持。同时,Flink 2.0 对 API 和配置进行了全面清理,移除了过时组件,为未来的发展奠定了坚实基础。感谢 165 位贡献者的辛勤付出,共同推动实时计算进入新纪元!
781 1
Apache Flink 2.0.0: 实时数据处理的新纪元
|
9月前
|
存储 SQL 人工智能
Apache Flink 2.0:Streaming into the Future
本文整理自阿里云智能高级技术专家宋辛童、资深技术专家梅源和高级技术专家李麟在 Flink Forward Asia 2024 主会场的分享。三位专家详细介绍了 Flink 2.0 的四大技术方向:Streaming、Stream-Batch Unification、Streaming Lakehouse 和 AI。主要内容包括 Flink 2.0 的存算分离云原生化、流批一体的 Materialized Table、Flink 与 Paimon 的深度集成,以及 Flink 在 AI 领域的应用。
1262 13
Apache Flink 2.0:Streaming into the Future
|
6月前
|
存储 大数据 数据处理
您有一份 Apache Flink 社区年度报告请查收~
您有一份 Apache Flink 社区年度报告请查收~
|
SQL 消息中间件 分布式计算
《Apache Flink 案例集(2022版)》——5.数字化转型——移动云Apache Flink 在移动云实时计算的实践(上)
《Apache Flink 案例集(2022版)》——5.数字化转型——移动云Apache Flink 在移动云实时计算的实践(上)
352 0
|
数据采集 分布式计算 Kubernetes
《Apache Flink 案例集(2022版)》——5.数字化转型——移动云Apache Flink 在移动云实时计算的实践(下)
《Apache Flink 案例集(2022版)》——5.数字化转型——移动云Apache Flink 在移动云实时计算的实践(下)
376 0
|
存储 SQL 传感器
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析2
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析2
769 0
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析2
|
SQL 消息中间件 分布式计算
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析1
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析1
450 0
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析1
|
SQL 消息中间件 分布式计算
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析3
【Flink】(04)Apache Flink 漫谈系列 —— 实时计算 Flink 与 Alibaba Cloud Realtime Compute 剖析3
230 0
|
27天前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队,全球领先的流计算引擎缔造者,支撑双11万亿级数据处理,推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才,地点覆盖北京、杭州、上海。技术深度参与开源核心,打造企业级实时计算解决方案,助力全球企业实现毫秒洞察。
356 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄

推荐镜像

更多