本文介绍在文件存储HDFS上搭建及使用Apache Flink的方法。

准备工作

在文件存储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配置文件。
    1. 执行如下命令打开hadoop-env.sh配置文件。
      vim /usr/local/hadoop-2.7.2/etc/hadoop/hadoop-env.sh
    2. 配置JAVA_HOME目录,如下所示。
      export JAVA_HOME=/usr/java/default
  3. 修改core-site.xml配置文件。
    1. 执行如下命令打开core-site.xml配置文件。
      vim /usr/local/hadoop-2.7.2/etc/hadoop/core-site.xml
    2. 在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配置文件。
    1. 执行如下命令打开mapred-site.xml配置文件。
      vim /usr/local/hadoop-2.7.2/etc/hadoop/mapred-site.xml
    2. 在mapred-site.xml配置文件中,配置如下信息。
      <configuration>
      <property>
            <name>mapreduce.framework.name</name>
            <value>yarn</value>
      </property>
      </configuration>
  5. 修改yarn-site.xml配置文件。
    1. 执行如下命令打开yarn-site.xml配置文件。
      vim /usr/local/hadoop-2.7.2/etc/hadoop/yarn-site.xml
    2. 在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配置文件。
    1. 执行如下命令打开slaves配置文件。
      vim /usr/local/hadoop-2.7.2/etc/hadoop/slaves
    2. 在slaves配置文件中,配置如下信息。
      node1
      node2
  7. 配置环境变量。
    1. 执行如下命令打开/etc/profile配置文件。
      vim /etc/profile
    2. 在/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
    3. 执行如下命令使配置生效。
      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
    flink-shade_1
    在依赖项中添加文件存储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>
    ...
    flink-shade_2
  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的挂载点,请根据您的实际情况替换。