# Flink Jar作业开发基础样例
#### 概述
用户可以基于Flink的API进行二次开发，构建自己的应用Jar包，提交到DLI队列运行，实现与MRS Kafka、HBase、Hive、HDFS，DWS，DCS等数据源的交互。
本章节以通过自定义作业与MRS进行交互为例进行说明。
更多样例代码请通过[DLI样例代码](https://uquery-sdk.obs-website.cn-north-1.myhuaweicloud.com/dli-example-code_1.0.0.zip)获取。
#### 环境准备
1. 登录MRS管理控制台，创建MRS集群，选择"开启kerberos"，勾选"Kafka", "HBase", "HDFS"等。请参见[《MapReduce服务用户指南》](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0513.html)的"购买自定义集群"的章节创建MRS。
2. "安全组规则"开通对应UDP/TCP端口。详细内容请参考[《私有云用户指南》](https://support.huaweicloud.com/usermanual-vpc/vpc_SecurityGroup_0001.html)中的"添加安全组规则"章节。
3. 进入MRS manager管理界面：
1. 创建机机账号，需确保该用户含有"hdfs_admin", "hbase_admin"权限，下载该用户认证凭据，其中包含"user.keytab" 和 "krb5.conf" 文件。
由于人机账号的keytab会随用户密码过期而失效，故建议使用机机账号进行配置。
2. 单击"服务管理"，下载客户端，单击"确定"。
3. 在MRS节点上下载配置文件，所需集群配置文件包含"hbase-site.xml"和"hiveclient.properties"。
4. 创建弹性资源池和队列。 弹性资源池与队列为DLI作业提供计算资源，[创建弹性资源池并添加队列](https://support.huaweicloud.com/usermanual-dli/dli_01_0505.html)。
5. 使用该DLI独享队列与MRS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)章节。
如何设置安全组规则，请参见[《虚拟私有云用户指南》](https://support.huaweicloud.com/usermanual-vpc/vpc_SecurityGroup_0001.html)中"安全组"章节。
6. 获取MRS集群全部节点的ip和域名映射，在DLI跨源连接修改主机信息中配置host映射。 如何添加IP域名映射，请参见《数据湖探索用户指南》中"[修改主机信息](https://support.huaweicloud.com/usermanual-dli/dli_01_0013.html)"章节。
Kafka服务端的端口如果监听在hostname上，则需要将Kafka Broker节点的hostname和IP的对应关系添加到DLI队列中。Kafka Broker节点的hostname和IP请联系Kafka服务的部署人员。
#### 前提条件
- 确保已创建弹性资源池和队列。
- 用户运行Flink Jar作业时，需要将二次开发的应用代码构建为Jar包，上传到已经创建的OBS桶中。并在DLI"数据管理"\>"程序包管理"页面创建程序包，具体请参考[创建程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。
DLI不支持下载功能，如果需要更新已上传的数据文件，可以将本地文件更新后重新上传。
- 由于DLI服务端已经内置了Flink的依赖包，并且基于开源社区版本做了安全加固。
为了避免依赖包兼容性问题或日志输出及转储问题，打包时请注意排除以下文件：
- 系统内置的依赖包，或者在Maven或者Sbt构建工具中将scope设为provided
- 日志配置文件（例如："log4j.properties"或者"logback.xml"等）
- 日志输出实现类JAR包（例如：log4j等）
- 上传自定义配置到OBS的桶必须为主账号下所创建的OBS桶。
- 使用Flink1.15版本的计算引擎时，需要用户自行配置委托，否则可能影响作业运行。 详细操作请参考[自定义DLI委托权限](https://support.huaweicloud.com/usermanual-dli/dli_01_0616.html)。
#### 使用方法
创建并提交Flink jar作业，详细操作步骤请参见《数据湖探索用户指南》中[创建Flink Jar作业](https://support.huaweicloud.com/usermanual-dli/dli_01_0457.html)章节。
1. 在DLI管理控制台的左侧导航栏中，单击"作业管理"\>"Flink作业"，进入"Flink作业"页面。
2. 在"Flink作业"页面右上角单击"新建作业"，弹出"新建作业"对话框。
图1新建Flink Jar作业
3. 配置作业信息。
表1作业配置信息
| 参数 | 参数说明 |
|:---|:---|
| 类型 | 选择Flink Jar。 |
| 名称 | 作业名称，只能由英文、中文、数字、中划线和下划线组成，并且长度为1～57字节。 说明： 作业名称必须是唯一的。 |
| 描述 | 作业的相关描述，且长度为0～512字节。 |
| 标签 | 使用标签标识云资源。包括"标签键"和"标签值"。如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在标签管理服务（TMS）中创建预定义标签。 具体请参考《[标签管理服务用户指南](https://support.huaweicloud.com/usermanual-tms/zh-cn_topic_0056266269.html)》。 说明： - 最多支持20个标签。 - 一个"键"只能添加一个"值"。 - 标签键：在输入框中输入标签键名称。 说明： - 标签键的最大长度为36个字符 ，不能包含"=","\*",",","\<","\>","\\","\|","/"，且首尾字符不能为空格。 - 若有预定义标签，可在输入框的下拉列表中进行选择。 - 标签值：在输入框中输入标签值。 说明： - 标签值的最大长度为43个字符，不能包含"=","\*",",","\<","\>","\\","\|","/"，且首尾字符不能为空格。 - 若有预定义标签，可在输入框的下拉列表中进行选择。 |
4. 单击"确定"，进入"编辑"页面。
5. 选择队列。Flink Jar作业只能运行在通用队列上。
图2选择队列
6. 上传Jar包。
Flink版本需要和用户Jar包指定的Flink版本保持一致。
图3上传Jar包
表2参数说明
| 名称 | 描述 |
|:---|:---|
| 应用程序 | 用户自定义的程序包。在选择程序包之前需要将对应的Jar包上传至OBS桶中，并在"数据管理\>程序包管理"中创建程序包，，具体请参考[创建程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 |
| 主类 | 指定加载的Jar包类名，如KafkaMessageStreaming。 - 默认：根据Jar包文件的Manifest文件指定。 - 指定：必须输入"类名"并确定类参数列表（参数间用空格分隔）。 说明： 当类属于某个包时，需携带包路径，例如：packagePath.KafkaMessageStreaming |
| 参数 | 指定类的参数列表，参数之间使用空格分隔。 |
| 依赖jar包 | 用户自定义的依赖程序包。在选择程序包之前需要将对应的Jar包上传至OBS桶中，并在"数据管理\>程序包管理"中创建程序包，包类型选择"jar"，具体请参考[创建程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 |
| 其他依赖文件 | 用户自定义的依赖文件。在选择依赖文件之前需要将对应的文件上传至OBS桶中，并在"数据管理\>程序包管理"中创建程序包，包类型没有限制，具体请参考[创建程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 通过在应用程序中添加以下内容可访问对应的依赖文件。其中，"fileName"为需要访问的文件名，"ClassName"为需要访问该文件的类名。 ``` ClassName.class.getClassLoader().getResource("userData/fileName") ``` |
| 作业特性 | 队列为CCE队列时，显示该参数。 - 基础型 - 自定义镜像：选择镜像名称和镜像版本。用户可在"容器镜像服务"设置的镜像。具体操作请参考《[容器镜像服务用户指南](https://support.huaweicloud.com/usermanual-swr/swr_01_0012.html)》。 |
| Flink版本 | 选择Flink版本前，需要先选择所属的队列。推荐选择Flink 1.15版本。 |
7. 配置作业参数。
Flink Jar作业最小化提交是指Flink仅提交作业必须的依赖项，而不是整个Flink环境。通过将非Connector的Flink依赖项（以flink-开头）和第三方库（如Hadoop、Hive、Hudi、Mysql-cdc）的作用域设置为provided，可以确保这些依赖项不会被包含在Jar作业中，从而实现最小化提交，避免依赖包与flink内核中依赖包冲突：
- 仅Flink 1.15版本支持Flink Jar作业最小化提交，通过在运行优化参数中配置flink.dli.job.jar.minimize-submission.enabled=true可以开启最小化提交。
- Flink相关依赖作用域请使用provided，即在依赖中添加\provided\。主要包含org.apache.flink组下以flink-开头的非Connector依赖。
- Hadoop、Hive、Hudi、Mysql-cdc相关依赖，作用域请使用provided，即在依赖中添加\provided\。
- Flink源代码中只有明确标注了@Public或者@PublicEvolving的才是公开供用户调用的方法，DLI只对这些方法的兼容性做出产品保证。
图4配置参数
表3参数说明
| 名称 | 描述 |
|:---|:---|
| CU数量 | 一个CU为1核4G的资源量。CU数量范围为2\~400个。 |
| 管理单元 | 设置管理单元的CU数，支持设置1\~4个CU，默认值为1个CU。 |
| 并行数 | 作业中每个算子的最大并行数。 说明： - 并行数不能大于计算单元（CU数量-管理单元CU数量）的4倍。 - 并行数最好大于用户作业里设置的并发数，否则有可能提交失败。 |
| TaskManager配置 | 用于设置TaskManager资源参数。 勾选后需配置下列参数： - "单TM所占CU数"：每个TaskManager占用的资源数量。 - "单TM Slot"：每个TaskManager包含的Slot数量。 |
| 保存作业日志 | 设置是否将作业运行时的日志信息保存到OBS。 勾选后需配置下列参数： "OBS桶"：选择OBS桶用于保存用户作业日志信息。如果选择的OBS桶是未授权状态，需要单击"OBS授权"。 |
| 作业异常告警 | 设置是否将作业异常告警信息，如作业出现运行异常或者欠费情况，以SMN的方式通知用户。 勾选后需配置下列参数： "SMN主题"： 选择一个自定义的SMN主题。如何自定义SMN主题，请参见[《消息通知服务用户指南》](https://support.huaweicloud.com/usermanual-smn/zh-cn_topic_0043961401.html)中"创建主题"章节。 |
| 异常自动重启 | 设置是否启动异常自动重启功能，当作业异常时将自动重启并恢复作业。 勾选后需配置下列参数： - "异常重试最大次数"：配置异常重试最大次数。单位为"次/小时"。 - 无限：无限次重试。 - 有限：自定义重试次数。 - "从Checkpoint恢复"：从最新保存的checkpoint恢复作业。 勾选该参数后，还需要选择"Checkpoint路径"。 "Checkpoint路径"：选择checkpoint保存路径。必须和应用程序中配置的Checkpoint地址相对应。且不同作业的路径不可一致，否则无法获取准确的Checkpoint。 |
8. 单击右上角"保存"，保存作业和相关参数。
9. 单击右上角"启动"，进入"启动Flink作业"页面，确认作业规格和费用，单击"立即启动"，启动作业。
启动作业后，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在"状态"列中可以查看作业状态。作业提交成功后，状态将由"提交中"变为"运行中"。运行完成后显示"已完成"。
如果作业状态为"提交失败"或"运行异常"，表示作业提交或运行失败。用户可以在作业列表中的"状态"列中，将鼠标移动到状态图标上查看错误信息，单击可以复制错误信息。根据错误信息解决故障后，重新提交。
其他功能按钮说明如下：
另存为：将新建作业另存为一个新作业。
#### 相关操作
- **怎样设置作业的参数？**
1. 在FLink作业列表中选择待编辑的作业。
2. 单击操作列"编辑"。
3. 在参数区域输入参数信息。 指定类的参数列表，参数之间使用空格分隔。
参数输入格式：--key1 value1 --key2 value2
例如：控制台入输入的参数
--bootstrap.server 192.168.168.xxx:9092
通过ParameterTool解析后的参数如下所示：
图5解析后的参数
- **怎样查看作业日志？**
1. 在FLink作业列表中单击作业名称，进入作业详情页面。
2. 单击"运行日志"，即可在控制台查看作业日志。 此处只展示最新的运行日志，更多信息请查看保存日志的OBS桶。