更新时间:2024-09-27 GMT+08:00

创建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 Jar作业时,需要将二次开发的应用代码构建为Jar包,上传到已经创建的OBS桶中。
  • 由于DLI服务端已经内置了Flink的依赖包,并且基于开源社区版本做了安全加固。为了避免依赖包兼容性问题或日志输出及转储问题,打包时请注意排除以下文件:
    • 系统内置的依赖包,或者在Maven或者Sbt构建工具中将scope设为provided
    • 日志配置文件(例如:“log4j.properties”或者“logback.xml”等)
    • 日志输出实现类JAR包(例如:log4j等)

注意事项

创建作业提交任务前,建议先开通云审计服务,用于记录与DLI服务相关的操作事件,便于日后的查询、审计和回溯。云审计服务支持的DLI操作列表详见使用CTS审计DLI服务

关于如何开通云审计服务以及如何查看追踪事件,请参考《云审计服务快速入门》。

创建Flink Jar作业

  1. 在DLI管理控制台的左侧导航栏中,单击作业管理>Flink作业,进入“Flink作业”页面。
  2. “Flink作业”页面右上角单击“新建作业”,弹出“新建作业”对话框。

    图1 新建Flink Jar作业

  3. 配置作业信息。

    表1 作业配置信息

    参数

    参数说明

    类型

    选择Flink Jar。

    名称

    作业名称,只能由英文、中文、数字、中划线和下划线组成,并且长度为1~57字节。

    说明:

    作业名称必须是唯一的。

    描述

    作业的相关描述,且长度为0~512字节。

    标签

    使用标签标识云资源。包括标签键和标签值。如果您需要使用同一标签标识多种云资源,即所有服务均可在标签输入框下拉选择同一标签,建议在标签管理服务(TMS)中创建预定义标签。

    如您的组织已经设定DLI的相关标签策略,则需按照标签策略规则为资源添加标签。标签如果不符合标签策略的规则,则可能会导致资源创建失败,请联系组织管理员了解标签策略详情。

    具体请参考《标签管理服务用户指南》。

    说明:
    • 最多支持20个标签。
    • 一个“键”只能添加一个“值”。
    • 每个资源中的键名不能重复。
    • 标签键:在输入框中输入标签键名称。
      说明:

      标签的键的最大长度为128个字符,标签的键可以包含任意语种字母、数字、空格和_ . : =+-@ ,但首尾不能含有空格,不能以_sys_开头。

    • 标签值:在输入框中输入标签值。
      说明:

      标签值的最大长度为255个字符,标签的值可以包含任意语种字母、数字、空格和_ . : =+-@ ,但首尾不能含有空格。

  4. 单击“确定”,进入编辑页面。
  5. 选择队列。
  6. 配置Flink Jar作业参数

    图2 配置Flink Jar作业参数
    表2 参数说明

    名称

    描述

    所属队列

    选择作业运行时使用的队列资源。

    应用程序

    选择已上传OBS桶中的自定义的程序包。

    主类

    指定加载的Jar包类名,如KafkaMessageStreaming。

    • 默认:根据Jar包文件的Manifest文件指定。
    • 指定:必须输入“类名”并确定类参数列表(参数间用空格分隔)。
    说明:

    当类属于某个包时,主类路径需要包含完整包路径,例如:packagePath.KafkaMessageStreaming

    参数

    指定类的参数列表,参数之间使用空格分隔。

    Flink参数支持非敏感的全局变量替换。例如,在“全局配置”>“全局变量”中新增全局变量windowsize,Flink Jar作业就可以添加参数-windowsSize {{windowsize}}。

    依赖jar包

    用户自定义的依赖程序包。依赖的相关程序包将会被放置到集群classpath下。

    在选择程序包之前需要将对应的Jar包上传至OBS桶中。打包Flink jar作业jar包时,为避免包信息冲突,不需要上传系统已有的内置依赖包。

    内置依赖包请参考DLI内置依赖包

    其他依赖文件

    用户自定义的依赖文件。其他依赖文件需要自行在代码中引用。

    在选择依赖文件之前需要将对应的文件上传至OBS桶中。

    通过在应用程序中添加以下内容可访问对应的依赖文件。其中,“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}}。

  7. 配置作业参数。

    图3 配置参数
    表3 参数说明

    名称

    描述

    CU数量

    一个CU为1核4G的资源量。CU数量范围为2~10000个。

    管理单元

    设置管理单元的CU数,支持设置1~4个CU,默认值为1个CU。

    并行数

    作业的并行数是指作业中各个算子的并行执行的子任务的数量,即算子子任务数就是其对应算子的并行度。

    说明:
    • 并行数不能大于计算单元(CU数量-管理单元CU数量)的4倍。
    • 并行数应大于用户作业里设置的并发数,否则有可能提交失败。

    TaskManager配置

    用于设置TaskManager资源参数。

    • 勾选后需配置下列参数:
      • “单TM所占CU数”:每个TaskManager占用的资源数量。
      • “单TM Slot”:每个TaskManager包含的Slot数量。
    • 不勾选该参数,,系统自动按照默认值为您配置。
      • “单TM所占CU数”:默认值为1。
      • “单TM Slot”:默认值为“(并行数 * 单TM所占CU数 )/(CU数量 - 管理单元)”。

    保存作业日志

    设置是否将作业运行时的日志信息保存到OBS桶。

    注意:

    该参数建议勾选,否则作业运行完成后不会生成运行日志,后续如果作业运行异常则无法获取运行日志进行定位。

    勾选后需配置下列参数:

    “OBS桶”:选择OBS桶用于保存用户作业日志信息。如果选择的OBS桶是未授权状态,需要单击“OBS授权”。

    作业异常告警

    设置是否将作业异常告警信息,如作业出现运行异常或者欠费情况,以SMN的方式通知用户。

    勾选后需配置下列参数:

    “SMN主题”

    选择一个自定义的SMN主题。如何自定义SMN主题,请参见《消息通知服务用户指南》“创建主题”章节。

    异常自动重启

    设置是否启动异常自动重启功能,当作业异常时将自动重启并恢复作业。

    勾选后需配置下列参数:

    • “异常重试最大次数”:配置异常重试最大次数。单位为“次/小时”。
      • 无限:无限次重试。
      • 有限:自定义重试次数。
    • “从Checkpoint恢复”:从保存的checkpoint恢复作业。

      勾选该参数后,还需要选择“Checkpoint路径”。

      “Checkpoint路径”:选择checkpoint保存路径。必须和应用程序中配置的Checkpoint地址相对应。且不同作业的路径不可一致,否则无法获取准确的Checkpoint。

  8. 单击右上角“保存”,保存作业和相关参数。
  9. 单击右上角“启动”,进入“启动Flink作业”页面,确认作业规格和费用,单击“立即启动”,启动作业,系统将自动跳转到Flink作业管理页面,新创建的作业将显示在作业列表中,在状态列中可以查看作业状态。

    • 作业提交成功后,状态将由提交中变为运行中。运行完成后显示“已完成”。
    • 如果作业状态为提交失败运行异常,表示作业提交或运行失败。用户可以在作业列表中的状态列中,将鼠标移动到状态图标上查看错误信息,单击可以复制错误信息。根据错误信息解决故障后,重新提交。