概述
用户可以基于Flink和Spark的API进行二次开发，构建自己的应用Jar包，提交到DLI队 列运行，实现与MRS Kafka、HBase、Hive、HDFS，DWS，DCS等数据源的交互。
本章节以通过自定义作业与MRS进行交互为例进行说明。更多样例代码请通过DLI样例 代码获取。
环境准备
1. 登录MRS管理控制台，创建MRS集群，选择“开启kerberos”，勾选“kafka”,
“hbase”, “hdfs”等。“安全组规则”开通对应UDP/TCP端口。
2. 进入MRS manager管理界面：
a. 创建机机账号，需确保该用户含有“hdfs_admin”, “hbase_admin”权 限，下载该用户认证凭据，其中包含“user.keytab” 和 “krb5.conf” 文 件。
说明
由于人机账号的keytab会随用户密码过期而失效，故建议使用机机账号进行配置。
b. 单击“服务管理”，下载客户端，单击“确定”。
c. 在MRS节点上下载配置文件，所需集群配置文件包含“hbase-site.xml”和
“hiveclient.properties”。
3. 创建DLI独享队列。
关于如何创建DLI独享队列，在购买队列时，选择“按需计费”，勾选“专属资源 模式”即可。具体操作请参见《数据湖探索用户指南》中创建队列章节。
4. 使用该DLI独享队列与MRS集群建立增强型跨源连接，且用户可以根据实际所需设 置相应安全组规则。
如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章 节。
如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。
5. 获取MRS集群全部节点的ip和域名映射，在DLI跨源连接修改主机信息中配置host 映射。
如何添加IP域名映射，请参见《数据湖探索用户指南》中“修改主机信息”章 节。
说明
Kafka服务端的端口如果监听在hostname上，则需要将Kafka Broker节点的hostname和IP 的对应关系添加到DLI队列中。Kafka Broker节点的hostname和IP请联系Kafka服务的部署 人员。
前提条件
● 确保已创建独享队列。
● 用户运行Flink Jar作业时，需要将二次开发的应用代码构建为Jar包，上传到已经 创建的OBS桶中。并在DLI“数据管理”>“程序包管理”页面创建程序包，具体 请参考创建程序包。
说明
DLI不支持下载功能，如果需要更新已上传的数据文件，可以将本地文件更新后重新上传。
● 由于DLI服务端已经内置了Flink的依赖包，并且基于开源社区版本做了安全加固。
为了避免依赖包兼容性问题或日志输出及转储问题，打包时请注意排除以下文 件：
– 系统内置的依赖包，或者在Maven或者Sbt构建工具中将scope设为provided – 日志配置文件（例如：“log4j.properties”或者“logback.xml”等）
– 日志输出实现类JAR包（例如：log4j等）
使用方法
创建并提交Flink jar作业，详细操作步骤请参见《数据湖探索用户指南》中创建Flink Jar作业章节。
步骤1 在DLI管理控制台的左侧导航栏中，单击“作业管理”>“Flink作业”，进入“Flink作 业”页面。
步骤2 在“Flink作业”页面右上角单击“新建作业”，弹出“新建作业”对话框。
图1-1 新建 Flink Jar 作业
步骤3 配置作业信息。
表1-5 作业配置信息
参数 参数说明
类型 选择Flink Jar。
名称 作业名称，只能由英文、中文、数字、中划线和下划线组成，并且长度 为1～57字节。
说明
作业名称必须是唯一的。
描述 作业的相关描述，且长度为0～512字节。
参数 参数说明
标签 使用标签标识云资源。包括“标签键”和“标签值”。如果您需要使用 同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一 标签，建议在标签管理服务（TMS）中创建预定义标签。具体请参考
《标签管理服务用户指南》。
说明
● 最多支持10个标签。
● 一个“键”只能添加一个“值”。
● 标签键：在输入框中输入标签键名称。
说明
– 标签键的最大长度为36个字符 ，不能包含“=”,“*”,“,”,“<”,“>”,
“\”,“|”,“/”，且首尾字符不能为空格。
– 若有预定义标签，可在输入框的下拉列表中进行选择。
● 标签值：在输入框中输入标签值。
说明
– 标签值的最大长度为43个字符，不能包含“=”,“*”,“,”,“<”,“>”,
“\”,“|”,“/”，且首尾字符不能为空格。
– 若有预定义标签，可在输入框的下拉列表中进行选择。
步骤4 单击“确定”，进入“编辑”页面。
步骤5 选择队列。Flink Jar作业只能运行在通用队列上。
说明
● Flink Jar作业只能运行在预先创建的独享队列上。
● 如果“所属队列”下拉框中无可用的独享队列，请先创建一个独享队列并将该队列绑定到当 前用户。
图1-2 选择队列
步骤6 上传Jar包。
图1-3 上传 Jar 包
表1-6 参数说明
名称 描述
应用程序 用户自定义的程序包。在选择程序包之前需要将对应的Jar包上传至 OBS桶中，并在“数据管理>程序包管理”中创建程序包，，具体请 参考创建程序包。
主类 指定加载的Jar包类名，如KafkaMessageStreaming。
● 默认：根据Jar包文件的Manifest文件指定。
● 指定：必须输入“类名”并确定类参数列表（参数间用空格分 隔）。
说明
当类属于某个包时，需携带包路径，例如：
packagePath.KafkaMessageStreaming
参数 指定类的参数列表，参数之间使用空格分隔。
依赖jar包 用户自定义的依赖程序包。在选择程序包之前需要将对应的Jar包上 传至OBS桶中，并在“数据管理>程序包管理”中创建程序包，包类 型选择“jar”，具体请参考创建程序包。
名称 描述
其他依赖文件 用户自定义的依赖文件。在选择依赖文件之前需要将对应的文件上 传至OBS桶中，并在“数据管理>程序包管理”中创建程序包，包类 型没有限制，具体请参考创建程序包。
通过在应用程序中添加以下内容可访问对应的依赖文件。其中，
“fileName”为需要访问的文件名，“ClassName”为需要访问该 文件的类名。
ClassName.class.getClassLoader().getResource("userData/fileName")
作业特性 队列为CCE队列时，显示该参数。
● 基础型
● 自定义镜像：选择镜像名称和镜像版本。用户可在“容器镜像服 务”设置的镜像。具体操作请参考《容器镜像服务用户指南》。
Flink版本 选择Flink版本前，需要先选择所属的队列。当前支持“1.10”和
“1.11”版本。
步骤7 配置作业参数。
图1-4 配置参数
表1-7 参数说明
名称 描述
CU数量 一个CU为1核4G的资源量。CU数量范围为2~400个。
管理单元 设置管理单元的CU数，支持设置1~4个CU，默认值为1个CU。
名称 描述
● “单TM Slot”：每个TaskManager包含的Slot数量。
保存作业日
说明
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/
maven-4.0.0.xsd">
<parent>
<artifactId>Flink-demo</artifactId>
<groupId>com.huaweicloud</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>flink-kafka-to-obs</artifactId>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<!--Flink 版本-->
<flink.version>1.10.0</flink.version>
<!--JDK 版本-->
<java.version>1.8</java.version>
<!--Scala 2.11 版本-->
<scala.binary.version>2.11</scala.binary.version>
<slf4j.version>2.13.3</slf4j.version>
<log4j.version>2.10.0</log4j.version>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<!-- flink -->