Flink Jar作业开发基础样例
概述
用户可以基于Flink的API进行二次开发，构建自己的应用Jar包，提交到DLI队列运行，实现与MRS Kafka、HBase、Hive、HDFS，DWS，DCS等数据源的交互。
本章节以通过自定义作业与MRS进行交互为例进行说明。
更多样例代码请通过DLI样例代码获取。
环境准备
- 登录MRS管理控制台，创建MRS集群，选择“开启kerberos”，勾选“Kafka”, “HBase”, “HDFS”等。请参见《MapReduce服务用户指南》的“购买自定义集群”的章节创建MRS。
- “安全组规则”开通对应UDP/TCP端口。详细内容请参考《私有云用户指南》中的“添加安全组规则”章节。
- 进入MRS manager管理界面：
- 创建机机账号，需确保该用户含有“hdfs_admin”, “hbase_admin”权限，下载该用户认证凭据，其中包含“user.keytab” 和 “krb5.conf” 文件。
由于人机账号的keytab会随用户密码过期而失效，故建议使用机机账号进行配置。
- 单击“服务管理”，下载客户端，单击“确定”。
- 在MRS节点上下载配置文件，所需集群配置文件包含“hbase-site.xml”和“hiveclient.properties”。
- 创建机机账号，需确保该用户含有“hdfs_admin”, “hbase_admin”权限，下载该用户认证凭据，其中包含“user.keytab” 和 “krb5.conf” 文件。
- 创建弹性资源池和队列。
- 使用该DLI独享队列与MRS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。
如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。
如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。
- 获取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的依赖包，并且基于开源社区版本做了安全加固。
- 上传自定义配置到OBS的桶必须为主账号下所创建的OBS桶。
- 使用Flink1.15版本的计算引擎时，需要用户自行配置委托，否则可能影响作业运行。
详细操作请参考自定义DLI委托权限。
使用方法
创建并提交Flink jar作业，详细操作步骤请参见《数据湖探索用户指南》中创建Flink Jar作业章节。
- 在DLI管理控制台的左侧导航栏中，单击>，进入“Flink作业”页面。
- 在“Flink作业”页面右上角单击“新建作业”，弹出“新建作业”对话框。
图1 新建Flink Jar作业
- 配置作业信息。
表1 作业配置信息 参数
参数说明
类型
选择Flink Jar。
名称
作业名称，只能由英文、中文、数字、中划线和下划线组成，并且长度为1～57字节。
说明：作业名称必须是唯一的。
描述
作业的相关描述，且长度为0～512字节。
标签
使用标签标识云资源。包括“标签键”和“标签值”。如果您需要使用同一标签标识多种云资源，即所有服务均可在标签输入框下拉选择同一标签，建议在标签管理服务（TMS）中创建预定义标签。
具体请参考《标签管理服务用户指南》。
说明：- 最多支持20个标签。
- 一个“键”只能添加一个“值”。
- 标签键：在输入框中输入标签键名称。
说明：
- 标签键的最大长度为36个字符 ，不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”，且首尾字符不能为空格。
- 若有预定义标签，可在输入框的下拉列表中进行选择。
- 标签值：在输入框中输入标签值。
说明：
- 标签值的最大长度为43个字符，不能包含“=”,“*”,“,”,“<”,“>”,“\”,“|”,“/”，且首尾字符不能为空格。
- 若有预定义标签，可在输入框的下拉列表中进行选择。
- 单击“确定”，进入页面。
- 选择队列。Flink Jar作业只能运行在通用队列上。
图2 选择队列
- 上传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版本。
- 配置作业参数。
图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。
- 单击右上角“保存”，保存作业和相关参数。
- 单击右上角“启动”，进入“启动Flink作业”页面，确认作业规格和费用，单击“立即启动”，启动作业。
启动作业后，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在列中可以查看作业状态。作业提交成功后，状态将由变为。运行完成后显示“已完成”。
如果作业状态为或，表示作业提交或运行失败。用户可以在作业列表中的列中，将鼠标移动到状态图标上查看错误信息，单击
可以复制错误信息。根据错误信息解决故障后，重新提交。
其他功能按钮说明如下：
另存为：将新建作业另存为一个新作业。
Java样例代码
- pom文件配置
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>com.huaweicloud</groupId> <artifactId>dli-flink-demo</artifactId> <version>1.0-SNAPSHOT</version> </parent> <groupId>org.example</groupId> <artifactId>flink-1.15-demo</artifactId> <properties> <flink.version>1.15.0</flink.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-statebackend-rocksdb</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.14.2</version> <scope>provided</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-assembly-plugin</artifactId> <version>3.3.0</version> <executions> <execution> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> <configuration> <archive> <manifest> <mainClass>com.huawei.dli.GetUserConfigFileDemo</mainClass> </manifest> </archive> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> </plugin> </plugins> <resources> <resource> <directory>src/main/resources</directory> <filtering>true</filtering> <includes> <include>**/*.*</include> </includes> </resource> </resources> </build> </project> - 示例代码
package com.huawei.dli; import com.huawei.dli.source.CustomParallelSource; import org.apache.flink.api.common.serialization.SimpleStringEncoder; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend; import org.apache.flink.core.fs.Path; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink; import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.net.URL; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.time.format.DateTimeFormatter; public class GetUserConfigFileDemo { private static final Logger LOG = LoggerFactory.getLogger(GetUserConfigFileDemo.class); public static void main(String[] args) { try { ParameterTool params = ParameterTool.fromArgs(args); LOG.info("Params: " + params.toString()); StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment(); // set checkpoint String checkpointPath = params.get("checkpoint.path", "obs://bucket/checkpoint/jobId_jobName/"); LocalDateTime localDateTime = LocalDateTime.ofEpochSecond(System.currentTimeMillis() / 1000, 0, ZoneOffset.ofHours(8)); String dt = localDateTime.format(DateTimeFormatter.ofPattern("yyyyMMdd_HH:mm:ss")); checkpointPath = checkpointPath + dt; streamEnv.setStateBackend(new EmbeddedRocksDBStateBackend()); streamEnv.getCheckpointConfig().setCheckpointStorage(checkpointPath); streamEnv.getCheckpointConfig().setExternalizedCheckpointCleanup( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); streamEnv.enableCheckpointing(30 * 1000); DataStream<String> stream = streamEnv.addSource(new CustomParallelSource()) .setParallelism(1) .disableChaining(); String outputPath = params.get("output.path", "obs://bucket/outputPath/jobId_jobName"); // Get user dependents config URL url = GetUserConfigFileDemo.class.getClassLoader().getResource("userData/user.config"); if (url != null) { Path filePath = org.apache.flink.util.FileUtils.absolutizePath(new Path(url.getPath())); try { String config = org.apache.flink.util.FileUtils.readFileUtf8(new File(filePath.getPath())); LOG.info("config is {}", config); // Do something by config } catch (IOException e) { LOG.error(e.getMessage(), e); } } // Sink OBS final StreamingFileSink<String> sinkForRow = StreamingFileSink .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8")) .withRollingPolicy(OnCheckpointRollingPolicy.build()) .build(); stream.addSink(sinkForRow); streamEnv.execute("sinkForRow"); } catch (Throwable e) { LOG.error(e.getMessage(), e); } } }
