创建Flink Jar作业
Flink Jar作业是基于Flink能力进行二次开发的场景，即构建自定义应用Jar包并提交到DLI的队列运行。
Flink Jar作业场景需要用户自行编写并构建应用Jar包，适用于对流计算处理复杂度要求较高的用户场景，且用户可以熟练掌握Flink二次开发能力。
本节操作介绍在DLI管理控制台创建Flink Jar作业的操作步骤。
前提条件
- 创建Flink Jar作业，访问其他外部数据源时，如访问OpenTSDB、HBase、Kafka、DWS、RDS、CSS、CloudTable、DCS Redis、DDS等，需要先创建跨源连接，打通作业运行队列到外部数据源之间的网络。
- 当前Flink作业支持访问的外部数据源详情请参考DLI常用跨源分析开发方式。
- 创建跨源连接操作请参见配置DLI与数据源网络连通（增强型跨源连接）。
创建完跨源连接后，可以通过“资源管理 > 队列管理”页面，单击“操作”列“更多”中的“测试地址连通性”，验证队列到外部数据源之间的网络连通是否正常。详细操作可以参考测试地址连通性。
- 用户运行Flink Jar作业时，需要将二次开发的应用代码构建为Jar包，上传到已经创建的OBS桶中。
- 由于DLI服务端已经内置了Flink的依赖包，并且基于开源社区版本做了安全加固。为了避免依赖包兼容性问题或日志输出及转储问题，打包时请注意排除以下文件：
- 系统内置的依赖包，或者在Maven或者Sbt构建工具中将scope设为provided
- 日志配置文件（例如：“log4j.properties”或者“logback.xml”等）
- 日志输出实现类JAR包（例如：log4j等）
注意事项
创建作业提交任务前，建议先开通云审计服务，用于记录与DLI服务相关的操作事件，便于日后的查询、审计和回溯。云审计服务支持的DLI操作列表详见使用CTS审计DLI服务。
关于如何开通云审计服务以及如何查看追踪事件，请参考《云审计服务快速入门》。
创建Flink Jar作业
- 在DLI管理控制台的左侧导航栏中，单击>，进入“Flink作业”页面。
- 在“Flink作业”页面右上角单击“新建作业”，弹出“新建作业”对话框。
图1 新建Flink Jar作业
- 配置作业信息。
表1 作业配置信息 参数
参数说明
类型
选择Flink Jar。
名称
作业名称，只能由英文、中文、数字、中划线和下划线组成，并且长度为1～57字节。
说明：作业名称必须是唯一的。
描述
作业的相关描述，且长度为0～512字节。
标签
使用标签标识云资源。包括标签键和标签值。如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在标签管理服务（TMS）中创建预定义标签。
如您的组织已经设定DLI的相关标签策略，则需按照标签策略规则为资源添加标签。标签如果不符合标签策略的规则，则可能会导致资源创建失败，请联系组织管理员了解标签策略详情。
具体请参考《标签管理服务用户指南》。
说明：- 最多支持20个标签。
- 一个“键”只能添加一个“值”。
- 每个资源中的键名不能重复。
- 标签键：在输入框中输入标签键名称。
说明：
标签的键的最大长度为128个字符，标签的键可以包含任意语种字母、数字、空格和_ . : =+-@ ,但首尾不能含有空格，不能以_sys_开头。
- 标签值：在输入框中输入标签值。
说明：
标签值的最大长度为255个字符，标签的值可以包含任意语种字母、数字、空格和_ . : =+-@ ,但首尾不能含有空格。
- 单击“确定”，进入编辑页面。
- 选择队列。
- 配置Flink Jar作业参数
图2 配置Flink Jar作业参数
表2 参数说明 名称
描述
所属队列
选择作业运行时使用的队列资源。
应用程序
选择Jar作业程序包。
Jar包的管理方式：
- 上传OBS管理程序包：提前将对应的jar包上传至OBS桶中。并在此处选择对应的OBS路径。
- 上传DLI管理程序包：提前将对应的jar包上传至OBS桶中，并在DLI管理控制台的中创建程序包，具体操作请参考创建DLI程序包。
Flink1.15推荐配置OBS中的程序包。不推荐使用DLI程序包，Flink1.15及以上版本将不再支持读取DLI程序包。
主类
指定加载的Jar包类名，如KafkaMessageStreaming。
- 默认：根据Jar包文件的Manifest文件指定。
- 指定：必须输入“类名”并确定类参数列表（参数间用空格分隔）。
说明：当类属于某个包时，主类路径需要包含完整包路径，例如：packagePath.KafkaMessageStreaming
参数
指定类的参数列表，参数之间使用空格分隔。
Flink参数支持非敏感的全局变量替换。例如，在“全局配置”>“全局变量”中新增全局变量windowsize，Flink Jar作业就可以添加参数-windowsSize {{windowsize}}。
依赖jar包
用户自定义的依赖程序包。依赖的相关程序包将会被放置到集群classpath下。
Jar包的管理方式：
- 上传OBS管理程序包：提前将对应的jar包上传至OBS桶中。并在此处选择对应的OBS路径。
- 上传DLI管理程序包：提前将对应的jar包上传至OBS桶中，并在DLI管理控制台的中创建程序包，具体操作请参考创建DLI程序包。
Flink1.15推荐配置OBS中的程序包。不推荐使用DLI程序包，Flink1.15及以上版本将不再支持读取DLI程序包。
打包Flink jar作业jar包时，为避免包信息冲突，不需要上传系统已有的内置依赖包。
内置依赖包请参考DLI内置依赖包。
其他依赖文件
用户自定义的依赖文件。其他依赖文件需要自行在代码中引用。
依赖文件的管理方式：
- 上传OBS管理程序包：提前将对应的依赖文件上传至OBS桶中。并在此处选择对应的OBS路径。
- 上传DLI管理程序包：提前将对应的依赖文件上传至OBS桶中，并在DLI管理控制台的中创建程序包，具体操作请参考创建DLI程序包。
Flink1.15推荐配置OBS中的程序包。不推荐使用DLI程序包，Flink1.15及以上版本将不再支持读取DLI程序包。
通过在应用程序中添加以下内容可访问对应的依赖文件。其中，“fileName”为需要访问的文件名，“ClassName”为需要访问该文件的类名。
ClassName.class.getClassLoader().getResource("userData/fileName")作业特性
队列为CCE队列时，显示该参数。
- 基础型
- 自定义镜像：选择镜像名称和镜像版本。用户可在“容器镜像服务”设置的镜像。具体操作请参考使用自定义镜像增强作业运行环境。
Flink版本
先选择所属的队列再选择Flink版本。
说明：不建议长期混用不同版本的Flink引擎。
- 长期混用不同版本的Flink引擎会导致代码在新旧版本之间不兼容，影响作业的执行效率。
- 当作业依赖于特定版本的库或组件，长期混用不同版本的Flink引擎可能会导致作业因依赖冲突而执行失败。
委托
使用Flink 1.15版本执行作业时，按需配置自定义委托用于授予DLI访问其他服务的操作权限。
优化参数
用户自定义的优化参数。参数格式为key=value。
Flink优化参数支持非敏感的全局变量替换。例如，在“全局配置”>“全局变量”中新增全局变量phase，Flink Jar作业就可以添加优化参数table.optimizer.agg-phase.strategy={{phase}}。
Flink 1.15版本支持Flink Jar作业最小化提交，通过在运行优化参数中配置flink.dli.job.jar.minimize-submission.enabled=true可以开启最小化提交功能。
说明：Flink Jar作业最小化提交是指Flink仅提交作业必须的依赖项，而不是整个Flink环境。通过将非Connector的Flink依赖项（以flink-开头）和第三方库（如Hadoop、Hive、Hudi、Mysql-cdc）的作用域设置为provided，可以确保这些依赖项不会被包含在Jar作业中，从而实现最小化提交，避免依赖包与flink内核中依赖包冲突：
- 仅Flink 1.15版本支持Flink Jar作业最小化提交。
- Flink相关依赖作用域请使用provided，即在依赖中添加<scope>provided</scope>。主要包含org.apache.flink组下以flink-开头的非Connector依赖。
- Hadoop、Hive、Hudi、Mysql-cdc相关依赖，作用域请使用provided，即在依赖中添加<scope>provided</scope>。
- Flink源代码中只有明确标注了@Public或者@PublicEvolving的才是公开供用户调用的方法，DLI只对这些方法的兼容性做出产品保证。
- 配置计算资源规格参数。
图3 配置参数
表3 参数说明 名称
描述
CU数量
一个CU为1核4G的资源量。CU数量范围为2~10000个。
管理单元
设置管理单元的CU数。
并行数
作业的并行数是指作业中各个算子的并行执行的子任务的数量，即算子子任务数就是其对应算子的并行度。
说明：- 并行数不能大于计算单元（CU数量-管理单元CU数量）的4倍。
- 并行数应大于用户作业里设置的并发数，否则有可能提交失败。
TaskManager配置
用于设置TaskManager资源参数。
- 勾选后需配置下列参数：
- “单TM所占CU数”：每个TaskManager占用的资源数量。
- “单TM Slot”：每个TaskManager包含的Slot数量。
- 不勾选该参数，，系统自动按照默认值为您配置。
- “单TM所占CU数”：默认值为1。
- “单TM Slot”：默认值为“（并行数 * 单TM所占CU数 ）/（CU数量 - 管理单元）”。
保存作业日志
设置是否将作业运行时的日志信息保存到OBS桶。
勾选“保存作业日志”时，还需要设置OBS桶信息。
注意：该参数建议勾选，否则作业运行完成后不会生成运行日志，后续如果作业运行异常则无法获取运行日志进行定位。
开启Checkpoint
Checkpoint用于定期保存作业状态，开启Checkpoint可以在系统故障时快速恢复指定的作业状态。
DLI开启Checkpoint有两种方式：
- 在作业代码中配置Checkpoint相关参数，适用于Flink 1.15及历史Flink版本。
- 在DLI管理控制台的Jar作业配置界面开启Checkpoint，适用于Flink 1.15及更高的引擎版本。
Flink 1.15版本请勿重复在作业代码和Jar作业配置界面配置Checkpoint相关参数，作业代码中的配置项优先级更高，重复配置可能导致作业在异常重启时使用错误的Checkpoint路径恢复数据，导致恢复失败或数据不一致。
勾选“开启Checkpoint”后配置以下参数开启Checkpoint：- Checkpoint间隔：Checkpoint的时间间隔，单位为秒。
- Checkpoint模式：选择Checkpoint的模式：
- At least once：事件至少被处理一次。
- Exactly once：事件仅被处理一次。
注意：- 勾选开启Checkpoint后需配置OBS桶参数， 用于保存Checkpoint信息，默认Checkpoint保存路径为：“桶名/jobs/checkpoint/作业ID开头的目录”。
- 开启Checkpoint后，请勿在作业代码中设置Checkpoint参数，作业代码中配置的参数优先级高于界面配置的参数优先级。重复配置可能导致作业在异常重启时使用错误的Checkpoint路径恢复数据，导致恢复失败或数据不一致。
- 开启Checkpoint后，如果同时勾选了“异常自动重启” ，并勾选了“从Checkpoint恢复”，无需再指定“Checkpoint路径”，系统会根据“开启Checkpoint”的配置信息自动指定。
OBS桶
勾选“保存作业日志”或“开启Checkpoint”时，该参数必选。
选择OBS桶用于保存用户作业日志信息。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。
作业异常告警
设置是否将作业异常告警信息，如作业出现运行异常或者欠费情况，以SMN的方式通知用户。
勾选后需配置下列参数：
“SMN主题”：
选择一个自定义的SMN主题。如何自定义SMN主题，请参见《消息通知服务用户指南》中“创建主题”章节。
异常自动重启
设置是否启动异常自动重启功能，当作业异常时将自动重启并恢复作业。
勾选后需配置下列参数：
- “异常重试最大次数”：配置异常重试最大次数。单位为“次/小时”。
- 无限：无限次重试。
- 有限：自定义重试次数。
- “从Checkpoint恢复”：从保存的checkpoint恢复作业。
勾选该参数后，需要选择“Checkpoint路径”。
“Checkpoint路径”：选择checkpoint保存路径。必须和应用程序中配置的Checkpoint地址相对应。且不同作业的路径不可一致，否则无法获取准确的Checkpoint。
说明：- 仅当同时勾选了“开启Checkpoint” 时无需再指定“Checkpoint路径”，系统会根据“开启Checkpoint”的配置信息自动指定。
- 如果未勾选“开启Checkpoint”，需要选择“Checkpoint路径”。
- 单击右上角“保存”，保存作业和相关参数。
- 单击右上角“启动”，进入“启动Flink作业”页面，确认作业规格和费用，单击“立即启动”，启动作业，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在列中可以查看作业状态。
- 作业提交成功后，状态将由变为。运行完成后显示“已完成”。
- 如果作业状态为或，表示作业提交或运行失败。用户可以在作业列表中的列中，将鼠标移动到状态图标上查看错误信息，单击
可以复制错误信息。根据错误信息解决故障后，重新提交。
