# 基于Flink JAR的数据处理 #### 操作场景 Flink Jar作业适用于需要自定义流处理逻辑、复杂的状态管理或特定库集成的数据分析场景。需要用户自行编写并构建Jar作业程序包，在提交Flink Jar作业前，将Jar作业程序包上传至OBS，并将程序包与数据和作业参数一起提交以运行作业。 本例介绍Flink Jar作业程序包运行的基本流程。 #### 操作流程 开始使用如下样例前，请务必按准备工作指导完成必要操作。 1. [规划并创建并行文件系统]：创建一个用于存储作业脚本的OBS并行文件系统。 2. [创建工作空间]：创建一个工作空间。 3. [创建运行作业的FlinkJob端点]：创建运行作业的Flink端点。 4. [使用API提交作业并查看运行结果]：使用API请求提交AI DataLake作业至Flink端点运行。 #### 准备工作 - 已注册账号并实名认证，且账号不能处于欠费或冻结状态。 - 已开通AI DataLake服务并授权使用云服务资源。 - 已开通OBS权限并进行了委托确认。 - 已经获取IAM Token，详细操作请参见[获取IAM用户Token](https://support.huaweicloud.com/api-iam/iam_30_0001.html)。 #### 步骤一：规划并创建并行文件系统 AI DataLake Flink通过OBS服务实现作业提交与数据存储，需要先在OBS控制台进行并行文件系统的创建，并导入样例数据或作业脚本。 1. [创建并行文件系统](https://support.huaweicloud.com/usermanual-obs/obs_13_0002.html)。 2. 在"并行文件系统"列表页面，单击已创建的并行文件系统名称。 3. 在"文件"页签，单击"上传文件"，将作业脚本上传至所创建的OBS并行文件系统中。 例如Flink Java代码"WindowJoin.java"，完整示例如下： ``` package org.apache.flink.streaming.examples.join; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.JoinFunction; import org.apache.flink.api.common.serialization.SimpleStringEncoder; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.connector.file.sink.FileSink; import org.apache.flink.core.fs.Path; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import java.time.Duration; public class WindowJoin { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); String outputPath = null; for (int i = 0; i < args.length - 1; i++) { if ("--output".equals(args[i])) { outputPath = args[i + 1]; break; } } DataStream> orderStream = env .fromElements( Tuple3.of("order1", 100, 1000L), Tuple3.of("order2", 200, 2000L), Tuple3.of("order3", 150, 3000L), Tuple3.of("order4", 300, 4000L) ) .assignTimestampsAndWatermarks( WatermarkStrategy .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) .withTimestampAssigner((event, ts) -> event.f2) ); DataStream> paymentStream = env .fromElements( Tuple3.of("order1", "credit", 1500L), Tuple3.of("order2", "debit", 2500L), Tuple3.of("order3", "credit", 3500L), Tuple3.of("order4", "debit", 4500L) ) .assignTimestampsAndWatermarks( WatermarkStrategy .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) .withTimestampAssigner((event, ts) -> event.f2) ); DataStream result = orderStream .join(paymentStream) .where(order -> order.f0) .equalTo(payment -> payment.f0) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .apply( (JoinFunction, Tuple3, String>) (order, payment) -> "orderId=" + order.f0 + ", amount=" + order.f1 + ", payMethod=" + payment.f1 ); if (outputPath != null) { FileSink sink = FileSink .forRowFormat(new Path(outputPath), new SimpleStringEncoder("UTF-8")) .withRollingPolicy(OnCheckpointRollingPolicy.build()) .build(); result.sinkTo(sink); } else { result.print(); } env.execute("WindowJoin QuickStart"); } } ``` Flink Java的pom.xml，完整示例如下： ``` 4.0.0 org.example window-join 1.0-SNAPSHOT jar 11 11 1.20.0 org.apache.flink flink-streaming-java ${flink.version} provided org.apache.flink flink-clients ${flink.version} provided org.apache.flink flink-connector-files ${flink.version} provided org.apache.maven.plugins maven-shade-plugin 3.5.0 package shade org.apache.flink.streaming.examples.join.WindowJoin ``` #### 步骤二：创建工作空间 1. 登录[AI DataLake管理控制台](https://console.huaweicloud.com/fabric/)。 2. 在左侧导航栏中，单击工作空间区域。 3. 在下拉列表中，单击"创建工作空间"。 4. 在"创建工作空间"页面，配置工作空间的相关参数。 表1工作空间的基本信息 | 类型 | 参数 | 是否必填 | 说明 | |:---|:---|:---|:---| | 基础信息 | 工作空间名称 | 是 | 工作空间的具体名称，同一个账号下的工作空间不可重名。 - 名称只能包含字母、数字、中划线、下划线。 - 名称字符长度为4\~32个字符。 工作空间名称不区分大小写，系统会自动转换为小写。 | | 基础信息 | 描述 | 否 | 对工作空间的简要描述，最长不超过255个字符。 | | 基础信息 | 企业项目 | 否 | 如果所建工作空间属于企业项目，可选择对应的企业项目。 企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理。 关于如何设置企业项目请参考《[企业管理用户指南](https://support.huaweicloud.com/usermanual-em/zh-cn_topic_0108763975.html)》。 说明： 只有开通了企业管理服务的用户才显示该参数。 | | 多模数据管理 | LakeFormation实例 | 是 | 选择当前工作空间关联的LakeFormation实例，每个工作空间需关联1个LakeFormation实例，空间创建后已关联的实例不支持修改。 - 选择已创建的LakeFormation实例，用于统一管理元数据和数据访问控制。 - 如无可用实例，请先创建LakeFormation实例，具体操作请参考[创建LakeFormation实例](https://support.huaweicloud.com/usermanual-lakeformation/lakeformation_03_00007.html)。 | 5. 确认所有配置信息无误。 单击"立即创建"按钮，完成工作空间创建。 工作空间创建成功后，您可以在工作空间管理的列表中查看新创建的工作空间。 #### 步骤三：创建运行作业的FlinkJob端点 1. 配置端点的基础信息。 表2端点的基础信息配置 | **参数** | **示例** | **参数说明** | |:---|:---|:---| | 端点类型 | FlinkJob | 选择端点类型为**FlinkJob。** | | 端点名称 | flink-job-01 | 端点名称创建后不可修改。 端点名称是系统内部的唯一标识，在API调用、CLI命令中引用该端点。 命名规则： - 名称只能包含小写字母、数字、中划线，且只能以字母开头，以字母或数字结尾。 - 输入长度不能超过63个字符。 例如：flink-endpoint-001、my-flink-job | | 端点显示名称 | 数据分析团队专用FlinkJob端点-01 | 端点显示名称创建后可修改。 端点显示名称用于用户界面展示，方便用户识别和记忆。 命名规则： - 名称只能输入中文、英文、数字、中划线。 - 输入长度不能超过63个字符。 例如：生产环境Flink引擎、数据分析师专用端点。 | | 添加描述 | 数据分析团队专用 | 填写描述信息，用于详细说明端点的用途、背景等信息，帮助其他用户理解该端点的创建目的和使用场景。 | 2. 配置端点的"资源配置"信息。 表3端点的资源配置 | 参数 | **示例** | 参数说明 | |:---|:---|:---| | 资源使用模式 | 预留资源 | 选择资源使用模式： - 预留资源：使用预留资源池，资源独享。适合负载稳定、持续运行的业务场景，如生产项目、关键任务。 - 混合模式：基线保障，自动扩容。优先消耗预留资源，高峰期自动触发弹性补位。适用于有规律波动的业务。 - 按需弹性：按任务实际运行时长扣费，无任务不产生费用。适合短期或偶发性的业务需求，如开发测试或临时任务。 关于计费有关详情请参考[计费概述](https://support.huaweicloud.com/price-aidatalake/aidatalake_12_0001.html)。 | 3. 配置端点的"存储配置"信息。 表4存储配置信息 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | 选择临时文件存储 | aidatalake-obs | 通过下拉菜单选择已有的并行文件系统。用于存储作业日志、临时数据、Checkpoint、Savepoint、HA数据等信息。 创建后不支持修改，若未配置，可单击"新建OBS并行文件系统"进行创建。 | | 委托名 | aidatalake-agency-job | 配置用于读写作业日志、临时数据、Checkpoint、Savepoint、HA数据等信息使用的委托。 通过下拉菜单选择已有的委托。创建后不支持修改，若未配置，可单击"新建委托"进行创建。具体操作请参考[创建自定义委托](https://support.huaweicloud.com/usermanual-aidatalake/aidatalake_01_0078.html)。 | 4. 网络配置。 在资源模式选择为"按需弹性"且工作空间没有创建网络时，需要创建网络并选择资源的网络信息。一个工作空间只支持创建一个网络。 5. 配置端点的"作业委托配置"信息。 表5作业委托配置信息 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | 作业委托配置 | aidatalake-agency-job | 委托给该端点的权限，会在运行时自动换取凭证并刷新至Flink集群。用户可在Flink集群中使用该委托凭证调用对应服务。使用Flink引擎执行作业时，需要您先在IAM页面创建相关委托，并在此处添加新建的委托信息。具体操作请参考[创建自定义委托](https://support.huaweicloud.com/usermanual-aidatalake/aidatalake_01_0078.html)。 | 6. 完成上述信息的配置后，单击"立即创建"，系统按照配置要求开始创建端点。 端点创建完成后，可在列表中查看相关信息，端点状态变为"运行中"后，即可提交作业到该端点中运行。 #### 步骤四：使用API提交作业并查看运行结果 1. 使用Postman等工具，调用运行作业API提交作业。 使用API提交作业请求示例如下： ``` POST /v2/workspaces/{workspace_id}/flink-jobs { "name": "flink-jar-job", "endpoint_name": "flink-job-01", "execution_mode": "STREAMING", "resource_config": { "parallelism": 1, "jobmanager_resource_spec": { "cpu": 500, "memory": 2048, "disk": 30 }, "taskmanager_resource_spec": { "cpu": 500, "memory": 2048, "disk": 30 } }, "job_config": { "type": "FLINK_JAR_JOB", "main_jar": "obs://aidatalake/jars/WindowJoin.jar", "main_class": "org.apache.flink.streaming.examples.join.WindowJoin", "main_args": [ "--output", "obs://aidatalake/jars/tmp/" ] }, "logging_config": { "root_logger_level": "INFO" } } ``` 2. 在OBS并行文件系统中找到obs://aidatalake/jars/tmp/2026-xx-xx--18/part-7efbxxxx-bbe7-xxxx-8482-ecaeca0cb6e9-0文件，查看JAR作业结果。 其结果参考如下： ``` orderId=order1, amount=100, payMethod=credit orderId=order4, amount=300, payMethod=debit orderId=order2, amount=200, payMethod=debit orderId=order3, amount=150, payMethod=credit ```