# 基于Flink SQL的数据处理 #### 操作场景 Flink作业支持使用其他服务作为数据源和数据输出进行数据实时计算操作。 本例采用Kafka服务作为数据源，以RDS作为数据输出，介绍创建并提交Flink OpenSource SQL作业进行实时计算的操作步骤。 #### 操作流程 开始使用如下样例前，请务必按准备工作指导完成必要操作。 1. [准备数据源]：创建一个Kafka实例作为数据源。 2. [准备数据输出]：创建一个RDS实例作为数据输出。 3. [规划并创建OBS并行文件系统]：创建一个用于存储作业脚本的OBS并行文件系统。 4. [创建工作空间]：创建一个工作空间。 5. [创建运行作业的FlinkJob端点]：创建运行作业的FlinkJob端点。 6. [使用API提交作业并查看运行结果]：使用API请求提交AI DataLake作业至Flink端点运行。 #### 准备工作 - 已注册账号并实名认证，且账号不能处于欠费或冻结状态。 - 已开通AI DataLake服务并授权使用云服务资源。 - 已开通LakeFormation服务并创建实例。 - 已开通Kafka数据接入服务。 - 已开通OBS权限并进行了委托确认。 - 已经获取IAM Token，详细操作请参见[获取IAM用户Token](https://support.huaweicloud.com/api-iam/iam_30_0001.html)。 #### 步骤一：准备数据源 本例以Kafka数据作为数据源。 1. 创建Kafka相关依赖资源。 在创建Kafka实例前您需要提前准备相关依赖资源，包括VPC、子网和安全组，并配置安全组。 - 创建VPC和子网的操作指导请参考[创建虚拟私有云和子网](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0013935842.html)，如果需要在已有VPC上创建和使用新的子网，请参考[为虚拟私有云创建新的子网](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0013748726.html)。 - 创建安全组的操作指导请参考[创建安全组](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0013748715.html)，为安全组添加规则的操作指导请参考[添加安全组规则](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0030969470.html)。 更多信息请参考《分布式消息服务Kafka用户指南》中的"[准备实例依赖资源](https://support.huaweicloud.com/usermanual-kafka/kafka-ug-180604012.html)"章节。 ![](https://support.huaweicloud.com/qs-aidatalake/public_sys-resources/note_3.0-zh-cn.png) - 创建的VPC与使用的Kafka服务应在相同的区域。 - 创建VPC和子网时，如无特殊需求，配置参数使用默认配置即可。 2. 开通Kafka数据接入服务，具体操作请参见[创建Kafka实例](https://support.huaweicloud.com/usermanual-kafka/kafka-ug-180604013.html)。 3. 单击购买的Kafka实例名称，单击"Topic管理 \> 创建Topic"，创建一个Topic。 #### 步骤二：准备数据输出 采用RDS作为数据输出，创建RDS MySQL实例。 1. 开通RDS MySQL实例。具体操作请参见[购买RDS for MySQL实例](https://support.huaweicloud.com/usermanual-rds-mysql/rds_10_0028.html)。 2. 登录MySQL，创建名称为flink的数据库。具体操作请参见[创建数据库](https://support.huaweicloud.com/usermanual-rds-mysql/rds_05_0019.html)。 3. 在flink库下创建orders表。 登录MySQL，单击"SQL窗口"，在SQL查询页面输入以下创建表语句，创建RDS MySQL表。 ``` CREATE TABLE `flink`.`orders` ( `order_id` VARCHAR(32) NOT NULL, `order_channel` VARCHAR(32) NULL, `order_time` VARCHAR(32) NULL, `pay_amount` DOUBLE UNSIGNED NOT NULL, `real_pay` DOUBLE UNSIGNED NULL, `pay_time` VARCHAR(32) NULL, `user_id` VARCHAR(32) NULL, `user_name` VARCHAR(32) NULL, `area_id` VARCHAR(32) NULL, PRIMARY KEY (`order_id`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci; ``` #### 步骤三：规划并创建OBS并行文件系统 AI DataLake Flink通过OBS服务实现作业提交与数据存储，需要先在OBS控制台进行并行文件系统的创建，并导入样例数据或作业脚本。 1. [创建并行文件系统](https://support.huaweicloud.com/usermanual-obs/obs_13_0002.html)。 2. 在"并行文件系统"列表页面，单击已创建的并行文件系统名称。 3. 在"文件"页签，单击"上传文件"，将作业脚本上传至所创建的OBS并行文件系统中。 例如Flink SQL脚本"kafkaSource.sql"，完整脚本示例如下： ``` CREATE TABLE kafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = 'testkafkatopic', -- 数据源kafka的topic，进入kafka控制台，单击已创建的Kafka实例名称，在Topic管理查看Topic名称 'properties.bootstrap.servers' = '192.168.x.xxx:xx92', -- 替换为kafka的内网连接地址和端口 'properties.group.id' = 'GroupId', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE TABLE jdbcSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://172.16.x.xxx:xx06/testrdsdb', -- testrdsdb为创建的RDS的数据库名，IP和端口替换为RDS MySQL的实例IP和端口 'table-name' = 'orders', 'username' = 'root', 'password' = '', 'sink.buffer-flush.max-rows' = '1' ); INSERT INTO jdbcSink SELECT * FROM kafkaSource; ``` #### 步骤四：创建工作空间 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-sql-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_SQL_JOB", "sql_script_file": "obs://aidatalake/sqls/kafkaSource.sql" }, "logging_config": { "root_logger_level": "INFO" } } ``` 2. 连接Kafka集群，向Kafka相应的topic中发送如下测试数据。 ``` {"order_id":"202607101000000001", "order_channel":"webShop", "order_time":"2026-07-10 10:00:00", "pay_amount":100.00, "real_pay":100.00, "pay_time":"2026-07-10 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202607101606060001", "order_channel":"appShop", "order_time":"2026-07-10 16:06:06", "pay_amount":200.00, "real_pay":180.00, "pay_time":"2026-07-10 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} ``` 3. 查看表中数据，在MySQL中执行sql查询语句。 ``` select * from orders; ``` 其结果参考如下： ``` 202607101000000001,webShop,2026-07-10 10:00:00,100,100,2026-07-10 10:02:03,0001,Alice,330106 202607101606060001,appShop,2026-07-10 16:06:06,200,180,2026-07-10 16:10:06,0001,Alice,330106 ```