# 创建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常用跨源分析开发方式](https://support.huaweicloud.com/usermanual-dli/dli_01_0410.html)。 - 创建跨源连接操作请参见[配置DLI与数据源网络连通（增强型跨源连接）](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)。 创建完跨源连接后，可以通过"资源管理 \> 队列管理"页面，单击"操作"列"更多"中的"测试地址连通性"，验证队列到外部数据源之间的网络连通是否正常。详细操作可以参考[测试队列与数据源网络连通性](https://support.huaweicloud.com/usermanual-dli/dli_01_0489.html)。 - 用户运行Flink Jar作业时，需要将二次开发的应用代码构建为Jar包，上传到已经创建的OBS桶中。 - 由于DLI服务端已经内置了Flink的依赖包，并且基于开源社区版本做了安全加固。为了避免依赖包兼容性问题或日志输出及转储问题，打包时请注意排除以下文件： - 系统内置的依赖包，或者在Maven或者Sbt构建工具中将scope设为provided - 日志配置文件（例如："log4j.properties"或者"logback.xml"等） - 日志输出实现类JAR包（例如：log4j等） #### 注意事项 创建作业提交任务前，建议先开通云审计服务，用于记录与DLI服务相关的操作事件，便于日后的查询、审计和回溯。云审计服务支持的DLI操作列表详见[使用CTS审计DLI服务](https://support.huaweicloud.com/usermanual-dli/dli_01_0318.html)。 关于如何开通云审计服务以及如何查看追踪事件，请参考《[云审计服务快速入门](https://support.huaweicloud.com/usermanual-cts/cts_02_0004.html)》。 #### 创建Flink Jar作业 1. 在DLI管理控制台的左侧导航栏中，单击"作业管理"\>"Flink作业"，进入"Flink作业"页面。 2. 在"Flink作业"页面右上角单击"新建作业"，弹出"新建作业"对话框。 图1新建Flink Jar作业 ![](https://support.huaweicloud.com/usermanual-dli/zh-cn_image_0000001254608837.png "点击放大") 3. 配置作业信息。 表1作业配置信息 | 参数 | 参数说明 | |:---|:---| | 类型 | 选择Flink Jar。 | | 名称 | 作业名称，只能由英文、中文、数字、中划线和下划线组成，并且长度为1～57字节。 说明： 作业名称必须是唯一的。 | | 描述 | 作业的相关描述，且长度为0～512字节。 | | 标签 | 使用标签标识云资源。包括标签键和标签值。如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在标签管理服务（TMS）中创建预定义标签。 如您的组织已经设定DLI的相关标签策略，则需按照标签策略规则为资源添加标签。标签如果不符合标签策略的规则，则可能会导致资源创建失败，请联系组织管理员了解标签策略详情。 具体请参考《[标签管理服务用户指南](https://support.huaweicloud.com/usermanual-tms/zh-cn_topic_0056266269.html)》。 说明： - 最多支持20个标签。 - 一个"键"只能添加一个"值"。 - 每个资源中的键名不能重复。 - 标签键：在输入框中输入标签键名称。 说明： 标签的键的最大长度为128个字符，标签的键可以包含任意语种字母、数字、空格和_ . : +-@ ,但首尾不能含有空格，不能以_sys_开头。 - 标签值：在输入框中输入标签值。 说明： 标签值的最大长度为255个字符，标签的值可以包含任意语种字母、数字、空格和_ . : +-@ 。 | 4. 单击"确定"，进入编辑页面。 5. 选择队列。 6. 配置Flink Jar作业参数 图2配置Flink Jar作业参数 ![](https://support.huaweicloud.com/usermanual-dli/zh-cn_image_0000001209849056.png "点击放大") 表2参数说明 | 名称 | 描述 | |:---|:---| | 所属队列 | 选择作业运行时使用的队列资源。 | | Flink版本 | Flink版本是选择作业运行时所使用的Flink的版本。不同版本的Flink支持不同的特性。 了解更多Flink版本的信息请参考DLI [Flink版本说明](https://support.huaweicloud.com/verspt-bulletin-dli/dli_bulletin_0012.html)。 选择使用Flink1.15版本时请在作业中配置允许DLI访问的云服务的委托信息。 Flink 1.15版本语法请参考[Flink OpenSource SQL1.15版本使用说明](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_15108.html)、[Flink OpenSource SQL1.15语法](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_15012.html)。 Flink 1.12版本语法请参考[Flink OpenSource SQL1.12语法](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0379.html)。 说明： 不建议长期混用不同版本的Flink引擎。 - 长期混用不同版本的Flink引擎会导致代码在新旧版本之间不兼容，影响作业的执行效率。 - 当作业依赖于特定版本的库或组件，长期混用不同版本的Flink引擎可能会导致作业因依赖冲突而执行失败。 | | 应用程序 | 选择Jar作业程序包。 Jar包的管理方式： - 上传OBS管理程序包：提前将对应的jar包上传至OBS桶中。并在此处选择对应的OBS路径。 - 上传DLI管理程序包：提前将对应的jar包上传至OBS桶中，并在DLI管理控制台的"数据管理\>程序包管理"中创建程序包，具体操作请参考[创建DLI程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 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程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 Flink1.15推荐配置OBS中的程序包，不推荐使用DLI程序包。Flink1.15以上版本将不再支持读取DLI程序包。 打包Flink jar作业jar包时，为避免包信息冲突，不需要上传系统已有的内置依赖包。 内置依赖包请参考[DLI内置依赖包](https://support.huaweicloud.com/usermanual-dli/dli_01_0397.html)。 | | 其他依赖文件 | 用户自定义的依赖文件。其他依赖文件需要自行在代码中引用。 依赖文件的管理方式： - 上传OBS管理程序包：提前将对应的依赖文件上传至OBS桶中。并在此处选择对应的OBS路径。 - 上传DLI管理程序包：提前将对应的依赖文件上传至OBS桶中，并在DLI管理控制台的"数据管理\>程序包管理"中创建程序包，具体操作请参考[创建DLI程序包](https://support.huaweicloud.com/usermanual-dli/dli_01_0367.html)。 Flink1.15推荐配置OBS中的程序包，不推荐使用DLI程序包。Flink1.15以上版本将不再支持读取DLI程序包。 通过在应用程序中添加以下内容可访问对应的依赖文件。其中，"fileName"为需要访问的文件名，"ClassName"为需要访问该文件的类名。 ``` ClassName.class.getClassLoader().getResource("userData/fileName") ``` | | 作业特性 | "作业特性"是配置创建Flink Jar时使用的镜像类型，用于指定DLI容器集群的镜像类型。 - 基础型：DLI提供的基础镜像。默认选择基础型。 - 自定义镜像：选择镜像名称和镜像版本。用户可在"容器镜像服务"设置的镜像。具体操作请参考[使用自定义镜像增强作业运行环境](https://support.huaweicloud.com/usermanual-dli/dli_01_0494.html)。 | | 委托 | 使用Flink 1.15及以上版本执行作业时，按需配置自定义委托用于授予DLI访问其他服务的操作权限。 自定义委托请参考[创建DLI自定义委托权限](https://support.huaweicloud.com/usermanual-dli/dli_01_0616.html)。 | | 优化参数 | 用户自定义的优化参数。参数格式为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，即在依赖中添加\provided\。主要包含org.apache.flink组下以flink-开头的非Connector依赖。 - Hadoop、Hive、Hudi、Mysql-cdc相关依赖，作用域请使用provided，即在依赖中添加\provided\。 - Flink源代码中只有明确标注了@Public或者@PublicEvolving的才是公开供用户调用的方法，DLI只对这些方法的兼容性做出产品保证。 | 7. 配置计算资源规格参数。 图3配置参数 ![](https://support.huaweicloud.com/usermanual-dli/zh-cn_image_0000001478636772.png "点击放大") 根据不同的Flink引擎版本，DLI提供了不同资源配置模板。 v2版本对比于V1模板不支持设置CU数量，支持直接设置Job Manager Memory和Task Manager Memory。 V1：适用于Flink 1.12、Flink 1.15 V2：适用于Flink 1.15、Flink 1.17 优先推荐使用V2版本的参数设置。 V1版本的具体参数说明请参考[表3]。 V2版本的具体参数说明请参考[表4]。 表3参数说明-V1 | 名称 | 描述 | |:---|:---| | CU数量 | 一个CU为1核4G的资源量。CU数量范围为2\~10000个。 说明： 当开启TaskManager配置时，为了优化弹性资源池队列的管理，在您设置"单TM Slot"后，为您自动调整CU数量与实际CU数量一致。 **CU数量=实际CU数量=max\[管理单元和TaskManager的CPU总和，(管理单元和TaskManager的内存总和/4)\]** - 管理单元和TaskManager的CPU总和=实际TM数 \* 单TM所占CU数 + 管理单元。 - 管理单元和TaskManager的内存总和= 实际TM数 \* 设置的单个TM的内存 + 管理单元内存 - 如果配置了单TM Slot数，实际TM数 = 并行数 / 单 TM Slot数。 - 如果没配置了单TM Slot数 ，实际TM数 = （CU数量 - 管理单元）/单TM所占CU数。 - 如果没在优化参数配置单个TM的内存和管理单元内存，默认单个TM的内存 = 单TM所占CU数 \* 4。管理单元内存 = 管理单元 \* 4。 - Spark资源并行度由Executor数量和Executor CPU核数共同决定。 | | 管理单元 | 设置管理单元的CU数。 如果当前作业是运行在基础版弹性资源池（16-64CUs）上，推荐管理单元的值不超过2，避免执行作业时资源调度失败。 | | 并行数 | 作业的并行数是指作业中各个算子的并行执行的子任务的数量，即算子的子任务数就是其对应算子的并行度。 说明： - 并行数不能大于计算单元（CU数量-管理单元CU数量）的4倍。 - 并行数应大于用户作业里设置的并发数，否则有可能提交失败。 | | TaskManager配置 | 用于设置TaskManager资源参数。 - 勾选后需配置下列参数： - "单TM所占CU数"：每个TaskManager占用的资源数量。 如果当前作业是运行在基础版弹性资源池（16-64CUs）上，推荐单TM所占CU数不超过2，避免执行作业时资源调度失败。 - "单TM Slot"：每个TaskManager包含的Slot数量。 - 不勾选该参数，系统自动按照默认值为您配置。 - "单TM所占CU数"：默认值为1。 - "单TM Slot"：默认值为"（并行数 \* 单TM所占CU数 ）/（CU数量 - 管理单元）"。 | 表4参数说明-V2 | 名称 | 描述 | |:---|:---| | 并行数 | 作业的并行数是指作业中各个算子的并行执行的子任务的数量，即算子的子任务数就是其对应算子的并行度。 说明： - 最小并行数不能小于1。默认值为1。 - 最大并行数不能大于计算单元（CU数量-管理单元）的4倍。 | | Job Manager CPU | 该参数用于设置JobManager可以使用的CPU核数。 Job Manager CPU默认值为1。最小值不能小于0.5。 如果当前作业是运行在基础版弹性资源池（16-64CUs）上，推荐Job Manager CPU的值不超过2，避免执行作业时资源调度失败。 | | Job Manager Memory | 该参数指用于设置JobManager可以使用的内存。 Job Manager Memory默认值为4G。最小值不能小于2G（不能小于2048M）。 单位默认GB，可设置为GB，MB。 | | Task Manager CPU | 该参数指用于设置TaskManager可以使用的CPU核数。 Task Manager CPU默认值为1。最小值不能小于0.5。 如果当前作业是运行在基础版弹性资源池（16-64CUs）上，推荐Task Manager CPU的值不超过2，避免执行作业时资源调度失败。 | | Task Manager Memory | 该参数指用于设置TaskManager可以使用的内存。 Task Manager Memory默认值4G。最小值不能小于2G（不能小于2048M）。 单位默认GB，可设置为GB，MB。 | | 单TM Slot | 该参数用于设置单个TaskManager可以提供的并行任务数量。每个 Task Slot 可以并行执行一个任务。增加 Task Slots 可以提高 TaskManager 的并行处理能力，但也会增加资源消耗。 Task Slots的数量与TaskManager的CPU数相关联，因为每个CPU可以提供一个Task Slot。 单TM Slot默认值为1。最小并行数不能小于1。 | Flink作业支持在Flink作业的参数配置中设置计算资源规格参数， 且自定义配置中的参数值优先级高于指定的值。 参数对应关系请参考[表5]。 ![](https://support.huaweicloud.com/usermanual-dli/public_sys-resources/note_3.0-zh-cn.png) Flink1.12优先推荐参考控制台的配置方法来配置计算资源规格参数，使用自定义配置参数可能会出现实际CUs统计数据不一致的问题。 表5控制台计算资源规格参数与Flink自定义配置中参数的对应关系 | 自定义配置 | 计算资源规格参数-V1 | 计算资源规格参数-V2 | 说明 | |:---|:---|:---|:---| | kubernetes.jobmanager.cpu | 管理单元 | Job Manager CPU | 该参数用于设置JobManager可以使用的CPU核数。 Job Manager CPU默认值为1。最小值不能小于0.5。 | | kubernetes.taskmanager.cpu | 单TM所占CU | Task Manager CPU | 该参数指用于设置TaskManager可以使用的CPU核数。 Task Manager CPU默认值为1。最小值不能小于0.5。 | | jobmanager.memory.process.size | - | Job Manager Memory | 该参数指用于设置JobManager可以使用的内存。 Job Manager Memory默认值为4G。最小值不能小于2G（不能小于2048M）。 单位默认GB，可设置为GB，MB。 | | taskmanager.memory.process.size | - | Task Manager Memory | 该参数指用于设置TaskManager可以使用的内存。 Task Manager Memory默认值4G。最小值不能小于2G（不能小于2048M）。 单位默认GB，可设置为GB，MB。 | 8. 配置日志和状态。 表6配置日志和状态 | 名称 | 描述 | |:---|:---| | 根日志级别 | Flink 1.15及以上版本支持配置该参数。 日志级别用于区分日志信息的重要性和紧急程度。根日志级别是全局的日志级别设置，它为整个应用程序提供了一个默认的日志级别。 当没有为特定的日志记录器或类设置日志级别时，根日志级别将被使用。例如，如果根日志级别设置为INFO，那么所有未单独设置级别的日志记录器都将输出INFO及以上级别的日志。 从低到高，日志级别的优先级通常为：TRACE\