本文介绍如何在Dataflow集群中运行Flink作业来消费OSS数据。

前提条件

  • 已开通E-MapReduce服务和OSS服务。
  • 已完成云账号的授权，详情请参见角色授权。

步骤一：准备环境

  1. 在本地安装Maven和Java环境。
    本示例中Maven是3.0版本，Java是1.8版本。
  2. 在E-MapReduce上创建Flink模式的Dataflow集群，详情请参见创建集群。
    本文以EMR-3.36.0版本的集群为例。

步骤二：准备测试数据

在创建Flink作业前，您需要在OSS上传测试数据。本示例上传一个test.txt文件，文件内容为Nothing is impossible for a willing heart. While there is a life, there is a hope~。

  1. 登录OSS管理控制台。
  2. 创建存储空间并上传测试数据文件，详情请参见创建存储空间和上传文件。
    测试数据的上传路径在后续步骤的代码中会使用，本例的上传路径为oss://emr-logs2/test/test.txt。
    说明 上传文件后，请保留OSS的登录窗口，后续仍会使用。

步骤三：制作JAR包并上传到OSS

本示例JAR包来源：下载E-MapReduce示例代码aliyun-emapreduce-demo，编译生成JAR包，上传至OSS中。

  1. 下载并解压缩aliyun-emapreduce-demo示例到本地。
  2. 在IntelliJ IDE中，单击file > open，打开解压缩后的aliyun-emapreduce-demo-master-2。
  3. 在下载文件中的pom.xml所在目录，执行如下命令制作JAR包。
    mvn clean package -DskipTests
  4. 返回OSS管理控制台。
  5. 上传JAR包至OSS任一路径下。
    JAR包的上传路径在后续步骤的代码中会使用，本示例的上传路径为oss://emr-logs2/test/examples-1.2.0.jar。

步骤四：创建并运行Flink作业

  1. 登录阿里云E-MapReduce控制台。
  2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
  3. 单击上方的数据开发页签。
  4. 在数据开发页面，创建项目，详情请参见项目管理。
  5. 进入新建的项目，新建Flink类型的作业，详情请参见Flink（VVR）作业配置。
  6. 新建Flink作业后，配置作业内容。
    作业内容示例如下：
    • Flink 1.10及之前版本
      run -m yarn-cluster -yjm 1024 -ytm 1024 -yn 4 -ys 4 -ynm flink-oss-sample -c com.aliyun.emr.example.flink.FlinkOSSSample ossref://emr-logs2/test/examples-1.2.0.jar --input oss://emr-logs2/test/test.txt
    • Flink 1.10之后版本
      run -m yarn-cluster -yjm 1024 -ytm 1024 -ys 4 -ynm flink-oss-sample -c com.aliyun.emr.example.flink.FlinkOSSSample ossref://emr-logs2/test/examples-1.2.0.jar --input oss://emr-logs2/test/test.txt
    示例代码中的关键参数说明如下：
    • ossref://emr-logs2/test/examples-1.2.0.jar：上传至OSS的JAR包。
    • oss://emr-logs2/test/test.txt：上传到OSS的测试数据。
  7. 作业配置完成后，单击右上方的运行。
    在运行作业对话框中，选择执行集群为新建的Flink模式的Dataflow集群。
  8. 单击确定。
    作业成功运行后，即成功实现了在E-MapReduce集群上运行Flink作业处理OSS数据。Flink作业结果

步骤五：查看作业提交日志和作业信息（可选）

如果需要定位作业失败的原因或了解作业的详细信息，则您可以查看作业的日志和作业信息。

  1. 查看作业提交日志。
    当前提交日志支持在E-MapReduce控制台查看，也支持在SSH客户端查看。
    • 提交作业后，您可以在E-MapReduce控制台的运行记录页签，单击待查看作业所在行的详情。Flink作业执行记录Flink作业日志
    • 通过SSH客户端登录到header节点，查看提交的日志信息。

      默认情况下，根据Flink的log4j配置（详情请参见/etc/ecm/flink-conf/log4j-yarn-session.properties），提交日志会保存在/mnt/disk1/log/flink/flink-{user}-client-{hostname}.log。

      其中，user为提交Flink作业的用户，hostname为提交作业所在的节点。以root用户在emr-header-1节点提交Flink作业为例，日志的路径为/mnt/disk1/log/flink/flink-flink-historyserver-0-emr-header-1.cluster-126601.log。

  2. 查看作业信息。
    通过Yarn UI可以查看Flink作业的信息。访问Yarn UI有SSH隧道和Knox两种方式，SSH隧道方式请参见通过SSH隧道方式访问开源组件Web UI，Knox方式请参见Knox和访问链接与端口。下面以Knox方式为例进行介绍。
    1. 在E-MapReduce控制台的访问链接与端口页面中，单击Yarn UI后的链接，
      YARN UI链接
    2. 在Hadoop控制台，单击作业的ID。
      查看作业运行详情。Hadoop控制台>Flink作业列表
      详细信息如下。Hadoop控制台>Flink作业详情
    3. 如果您需要查看运行中的Flink作业，则可以在作业详情页面，单击Tracking URL后面的链接，进入Flink Dashboard查看。
    4. 作业运行结束后，通过访问http://emr-header-1:8082，您可以查看所有已经完成的作业列表。