本文分享自华为云社区《【云小课】EI第47课 MRS离线数据分析-通过Flink作业处理OBS数据》，作者：Hello EI 。

MRS 支持在大数据存储容量大、计算资源需要弹性扩展的场景下，用户将数据存储在 OBS 服务中，使用 MRS 集群仅做数据计算处理的存算分离模式。

Flink 是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。

本文将向您介绍如何在 MRS 集群中运行 Flink 作业来处理 OBS 中存储的数据。

Flink 最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发 pipeline 处理数据，时延毫秒级，且兼具可靠性。

在本示例中，我们使用 MRS 集群内置的 Flink WordCount 作业程序，来分析 OBS 文件系统中保存的源数据，以统计源数据中的单词出现次数。

当然您也可以获取 MRS 服务样例代码工程，参考 Flink 开发指南开发其他 Flink 流作业程序。

本案例基本操作流程如下所示：

创建 MRS 集群

创建并购买一个包含有 Flink 组件的 MRS 集群，详情请参见购买自定义集群。

本文以购买 MRS 3.1.0 版本的集群为例，集群未开启 Kerberos 认证。

在本示例中，由于我们要分析处理 OBS 文件系统中的数据，因此在集群的高级配置参数中要为 MRS 集群绑定 IAM 权限委托，使得集群内组件能够对接 OBS 并具有对应文件系统目录的操作权限。

您可以直接选择系统默认的“MRS_ECS_DEFAULT_AGENCY”，也可以自行创建其他具有 OBS 文件系统操作权限的自定义委托。

集群购买成功后，在 MRS 集群的任一节点内，使用 omm 用户安装集群客户端，具体操作可参考安装并使用集群客户端。

例如客户端安装目录为“/opt/client”。

准备测试数据

在创建 Flink 作业进行数据分析前，我们需要在提前准备待分析的测试数据，并将该数据上传至 OBS 文件系统中。

1、本地创建一个“mrs_flink_test.txt”文件，例如文件内容如下：

This is a test demo for MRS Flink. Flink is a unified computing framework that supports both batch processing and stream processing. It provides a stream data processing engine that supports data distribution and parallel computing.

2、在云服务列表中选择“存储 > 对象存储服务”，登录 OBS 管理控制台。

3、单击“并行文件系统”，创建一个并行文件系统，并上传测试数据文件。

例如创建的文件系统名称为“mrs-demo-data”，单击系统名称，在“文件”页面中，新建一个文件夹“flink”，上传测试数据至该目录中。

则本示例的测试数据完整路径为“obs://mrs-demo-data/flink/mrs_flink_test.txt”。

4、上传数据分析应用程序。

使用管理台界面直接提交作业时，将已开发好的 Flink 应用程序 jar 文件也可以上传至 OBS 文件系统中，或者 MRS 集群内的 HDFS 文件系统中。

本示例中我们使用 MRS 集群内置的 Flink WordCount 样例程序，可从 MRS 集群的客户端安装目录中获取，即“/opt/client/Flink/flink/examples/batch/WordCount.jar”。

将“WordCount.jar”上传至“mrs-demo-data/program”目录下。

创建并运行 Flink 作业

方式 1：在控制台界面在线提交作业。

1. 登录 MRS 管理控制台，单击 MRS 集群名称，进入集群详情页面。

2. 在集群详情页的“概览”页签，单击“IAM 用户同步”右侧的“单击同步”进行 IAM 用户同步。

3. 单击“作业管理”，进入“作业管理”页签。

4. 单击“添加”，添加一个 Flink 作业。作业类型：Flink 作业名称：自定义，例如 flink_obs_test。执行程序路径：本示例使用 Flink 客户端的 WordCount 程序为例。运行程序参数：使用默认值。执行程序参数：设置应用程序的输入参数，“input”为待分析的测试数据，“output”为结果输出文件。

例如本示例中，我们设置为“--input obs://mrs-demo-data/flink/mrs_flink_test.txt --output obs://mrs-demo-data/flink/output”。

• 服务配置参数：使用默认值即可，如需手动配置作业相关参数，可参考运行Flink作业。

5.确认作业配置信息后，单击“确定”，完成作业的新增，并等待运行完成。

方式 2：通过集群客户端提交作业。

1、使用 root 用户登录集群客户端节点，进入客户端安装目录。

su - ommcd /opt/clientsource bigdata_env

2、执行以下命令验证集群是否可以访问 OBS。

hdfs dfs -ls obs://mrs-demo-data/flink

3、提交 Flink 作业，指定源文件数据进行消费。

flink run -m yarn-cluster /opt/client/Flink/flink/examples/batch/WordCount.jar --input obs://mrs-demo-data/flink/mrs_flink_test.txt --output obs://mrs-demo/data/flink/output2

执行后结果类似如下：

...
Cluster started: Yarn cluster with application id application_1654672374562_0011
Job has been submitted with JobID a89b561de5d0298cb2ba01fbc30338bc
Program execution finished
Job with JobID a89b561de5d0298cb2ba01fbc30338bc has finished.
Job Runtime: 1200 ms

查看作业执行结果

1、作业提交成功后，登录 MRS 集群的 FusionInsight Manager 界面，选择“集群 > 服务 > Yarn”。

2、单击“ResourceManager WebUI”后的链接进入 Yarn Web UI 界面，在 Applications 页面查看当前 Yarn 作业的详细运行情况及运行日志。

3、等待作业运行完成后，在 OBS 文件系统中指定的结果输出文件中可查看数据分析输出的结果。

下载“output”文件到本地并打开，可查看输出的分析结果。

a 3
and 2
batch 1
both 1
computing 2
data 2
demo 1
distribution 1
engine 1
flink 2
for 1
framework 1
is 2
it 1
mrs 1
parallel 1
processing 3
provides 1
stream 2
supports 2
test 1
that 2
this 1
unified 1

使用集群客户端命令行提交作业时，若不指定输出目录，在作业运行界面也可直接查看数据分析结果。

Job with JobID xxx has finished.
Job Runtime: xxx ms
Accumulator Results:
- e6209f96ffa423974f8c7043821814e9 (java.util.ArrayList) [31 elements]
(a,3)
(and,2)
(batch,1)
(both,1)
(computing,2)
(data,2)
(demo,1)
(distribution,1)
(engine,1)
(flink,2)
(for,1)
(framework,1)
(is,2)
(it,1)
(mrs,1)
(parallel,1)
(processing,3)
(provides,1)
(stream,2)
(supports,2)
(test,1)
(that,2)
(this,1)
(unified,1)

点击关注，第一时间了解华为云新鲜技术~