更新时间:2024-06-13 GMT+08:00

Flink Jar作业开发基础样例

概述

用户可以基于Flink的API进行二次开发,构建自己的应用Jar包,提交到DLI队列运行,实现与MRS Kafka、HBase、Hive、HDFS,DWS,DCS等数据源的交互。

本章节以通过自定义作业与MRS进行交互为例进行说明。

更多样例代码请通过DLI样例代码获取。

环境准备

  1. 登录MRS管理控制台,创建MRS集群,选择“开启kerberos”,勾选“Kafka”, “HBase”, “HDFS”等。请参见《MapReduce服务用户指南》的“购买自定义集群”的章节创建MRS。
  2. “安全组规则”开通对应UDP/TCP端口。详细内容请参考《私有云用户指南》中的“添加安全组规则”章节。
  3. 进入MRS manager管理界面:
    1. 创建机机账号,需确保该用户含有“hdfs_admin”, “hbase_admin”权限,下载该用户认证凭据,其中包含“user.keytab”“krb5.conf” 文件。

      由于人机账号的keytab会随用户密码过期而失效,故建议使用机机账号进行配置。

    2. 单击“服务管理”,下载客户端,单击“确定”
    3. 在MRS节点上下载配置文件,所需集群配置文件包含“hbase-site.xml”“hiveclient.properties”
  4. 创建弹性资源池和队列。

    弹性资源池与队列为DLI作业提供计算资源,创建弹性资源池弹性资源池添加队列

  5. 使用该DLI独享队列与MRS集群建立增强型跨源连接,且用户可以根据实际所需设置相应安全组规则。

    如何建立增强型跨源连接,请参见《数据湖探索用户指南》增强型跨源连接章节。

    如何设置安全组规则,请参见《虚拟私有云用户指南》“安全组”章节。

  6. 获取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等)
  • 上传自定义配置到OBS的桶必须为主账号下所创建的OBS桶。
  • 使用Flink1.15版本的计算引擎时,需要用户自行配置委托,否则可能影响作业运行。

    详细操作请参考自定义DLI委托权限

使用方法

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

    图1 新建Flink Jar作业

  3. 配置作业信息。

    表1 作业配置信息

    参数

    参数说明

    类型

    选择Flink Jar。

    名称

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

    说明:

    作业名称必须是唯一的。

    描述

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

    标签

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

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

    说明:
    • 最多支持20个标签。
    • 一个“键”只能添加一个“值”。
    • 标签键:在输入框中输入标签键名称。
      说明:
      • 标签键的最大长度为36个字符 ,不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”,且首尾字符不能为空格。
      • 若有预定义标签,可在输入框的下拉列表中进行选择。
    • 标签值:在输入框中输入标签值。
      说明:
      • 标签值的最大长度为43个字符,不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”,且首尾字符不能为空格。
      • 若有预定义标签,可在输入框的下拉列表中进行选择。

  4. 单击“确定”,进入编辑页面。
  5. 选择队列。Flink Jar作业只能运行在通用队列上。

    图2 选择队列

  6. 上传Jar包。

    Flink版本需要和用户Jar包指定的Flink版本保持一致。
    图3 上传Jar包
    表2 参数说明

    名称

    描述

    应用程序

    用户自定义的程序包。在选择程序包之前需要将对应的Jar包上传至OBS桶中,并在数据管理>程序包管理中创建程序包,,具体请参考创建程序包

    主类

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

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

    当类属于某个包时,需携带包路径,例如:packagePath.KafkaMessageStreaming

    参数

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

    依赖jar包

    用户自定义的依赖程序包。在选择程序包之前需要将对应的Jar包上传至OBS桶中,并在数据管理>程序包管理中创建程序包,包类型选择“jar”,具体请参考创建程序包

    其他依赖文件

    用户自定义的依赖文件。在选择依赖文件之前需要将对应的文件上传至OBS桶中,并在数据管理>程序包管理中创建程序包,包类型没有限制,具体请参考创建程序包

    通过在应用程序中添加以下内容可访问对应的依赖文件。其中,“fileName”为需要访问的文件名,“ClassName”为需要访问该文件的类名。

    ClassName.class.getClassLoader().getResource("userData/fileName")

    作业特性

    队列为CCE队列时,显示该参数。

    • 基础型
    • 自定义镜像:选择镜像名称和镜像版本。用户可在“容器镜像服务”设置的镜像。

    Flink版本

    选择Flink版本前,需要先选择所属的队列。推荐选择Flink 1.15版本。

  7. 配置作业参数。

    图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主题,请参见《消息通知服务用户指南》“创建主题”章节。

    异常自动重启

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

    勾选后需配置下列参数:

    • “异常重试最大次数”:配置异常重试最大次数。单位为“次/小时”。
      • 无限:无限次重试。
      • 有限:自定义重试次数。
    • “从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桶。