最新动态
功能总览
服务公告
- 产品公告
- 版本支持公告
产品介绍
- 图解数据湖探索
- 数据湖探索简介
- 产品优势
- 应用场景
- 约束与限制
- 产品规格
- 安全
- 权限管理
- 配额管理
- 与其他云服务的关系
- 基本概念
计费说明
- 计费概述
- 计算计费
  - 弹性资源池计费
- 存储计费
- 扫描量计费
- 套餐包计费
- 计费样例
- 续费
- 费用账单
- 欠费说明
- 停止计费
- 计费相关问题
- 修订记录
快速入门
- 使用DLI提交SQL作业查询OBS数据
- 使用DLI提交SQL作业查询RDS MySQL数据
- 使用DLI提交Flink OpenSource SQL作业查询RDS MySQL数据
- 使用DLI提交Flink Jar作业
- 使用DLI提交Spark Jar作业
- 入门实践
用户指南
- DLI作业开发流程
- 准备工作
- 创建弹性资源池和队列
- 创建数据库和表
- 数据迁移与数据传输
- 配置DLI访问其他云服务的委托权限
- 使用DLI提交SQL作业
- 使用DLI提交Flink作业
- 使用DLI提交Spark作业
- 使用Notebook实例提交DLI作业
- 使用CES监控DLI服务
- 使用CTS审计DLI服务
- 权限管理
- DLI常用管理操作
最佳实践
- 最佳实践内容概览
- 使用DLI分析车联网场景驾驶行为数据
- 使用DLI将CSV数据转换为Parquet数据
- 使用DLI分析电商BI报表
- 使用DLI分析账单消费数据
- 使用DLI分析电商实时业务数据
- 使用BI工具连接DLI分析数据
开发指南
- 使用客户端工具连接DLI
- SQL作业开发指南
- Flink作业开发指南
- Spark Jar作业开发指南
Spark SQL语法参考
- Spark SQL常用配置项说明
- Spark SQL语法概览
- Spark开源命令支持说明
- 数据库相关
- 表相关
- 数据相关
- 导出查询结果
- 跨源连接相关
- 视图相关
  - 创建视图
  - 删除视图
- 查看计划
- 数据权限相关
- 数据类型
- 自定义函数
- 内置函数
  - 日期函数
    - 日期函数概览
    - add_months
    - current_date
    - current_timestamp
    - date_add
    - dateadd
    - date_sub
    - date_format
    - datediff
    - datediff1
    - datepart
    - datetrunc
    - day/dayofmonth
    - from_unixtime
    - from_utc_timestamp
    - getdate
    - hour
    - isdate
    - last_day
    - lastday
    - minute
    - month
    - months_between
    - next_day
    - quarter
    - second
    - to_char
    - to_date
    - to_date1
    - to_utc_timestamp
    - trunc
    - unix_timestamp
    - weekday
    - weekofyear
    - year
  - 字符串函数
    - 字符串函数概览
    - ascii
    - concat
    - concat_ws
    - char_matchcount
    - encode
    - find_in_set
    - get_json_object
    - instr
    - instr1
    - initcap
    - keyvalue
    - length
    - lengthb
    - levenshtein
    - locate
    - lower/lcase
    - lpad
    - ltrim
    - parse_url
    - printf
    - regexp_count
    - regexp_extract
    - replace
    - regexp_replace
    - regexp_replace1
    - regexp_instr
    - regexp_substr
    - repeat
    - reverse
    - rpad
    - rtrim
    - soundex
    - space
    - substr/substring
    - substring_index
    - split_part
    - translate
    - trim
    - upper/ucase
  - 数学函数
    - 数学函数概览
    - abs
    - acos
    - asin
    - atan
    - bin
    - bround
    - cbrt
    - ceil
    - conv
    - cos
    - cot1
    - degrees
    - e
    - exp
    - factorial
    - floor
    - greatest
    - hex
    - least
    - ln
    - log
    - log10
    - log2
    - median
    - negative
    - percentlie
    - percentlie_approx
    - pi
    - pmod
    - positive
    - pow
    - radians
    - rand
    - round
    - shiftleft
    - shiftright
    - shiftrightunsigned
    - sign
    - sin
    - sqrt
    - tan
  - 聚合函数
    - 聚合函数概览
    - avg
    - corr
    - count
    - covar_pop
    - covar_samp
    - max
    - min
    - percentile
    - percentile_approx
    - stddev_pop
    - stddev_samp
    - sum
    - variance/var_pop
    - var_samp
  - 分析窗口函数
    - 分析窗口函数概览
    - cume_dist
    - first_value
    - last_value
    - lag
    - lead
    - percent_rank
    - rank
    - row_number
  - 其他函数
    - 函数概览
    - decode1
    - javahash
    - max_pt
    - ordinal
    - trans_array
    - trunc_numeric
    - url_decode
    - url_encode
- SELECT
  - 基本语句
  - 排序
  - 分组
  - 连接
  - 子句
    - FROM
    - OVER
    - WHERE
    - HAVING
    - 多层嵌套子查询
  - 别名SELECT
    - 表别名
    - 列别名
  - 集合运算SELECT
    - UNION
    - INTERSECT
    - EXCEPT
  - WITH...AS
  - CASE...WHEN
    - 简单CASE函数
    - CASE搜索函数
- 标示符
- 运算符
Flink SQL语法参考
- Flink Opensource SQL1.15语法参考
- Flink Opensource SQL1.12语法参考
- Flink Opensource SQL1.10语法参考
HetuEngine SQL语法参考
- HetuEngine SQL语法
- 数据类型隐式转换
  - 简介
  - 隐式转换对照表
- 附录
  - 本文样例表数据准备
  - 常用数据源语法兼容性
Hudi SQL语法参考
- Hudi表概述
- DLI Hudi元数据
- DLI Hudi开发规范
- DLI中使用Hudi开发作业
- DLI Hudi SQL语法参考
- Spark datasource API语法参考
  - API语法说明
  - Hudi锁配置说明
- 数据管理维护
- Hudi常见配置参数
Delta SQL语法参考
- DLI Delta表概述
- DLI中使用Delta开发作业
  - DLI Delta元数据
  - 在DLI使用Delta提交Spark Jar作业
- Delta Time Travel
- Delta清理和优化
- Delta SQL语法参考
- Delta常见配置参数
- DLI Delta常见问题
API参考
- API使用前必读
  - 概述
  - 调用说明
  - 终端节点
  - 约束与限制
  - 基本概念
- API概览
- 如何调用API
  - 构造请求
  - 认证鉴权
  - 返回结果
- API快速入门
  - 创建并提交SQL作业
  - 创建并提交Spark作业
  - 创建并提交Flink作业
  - 创建并使用跨源链接
- 权限相关API
  - 数据赋权（用户或项目）
  - 查看赋权对象使用者权限信息
- 全局变量相关API
  - 创建全局变量
  - 删除全局变量
  - 修改全局变量
  - 查询所有全局变量
- 资源标签相关API
  - 批量添加资源标签
  - 批量删除资源标签
  - 查询资源实例数量
  - 查询资源实例列表
  - 查询指定资源类型的标签信息
  - 查询指定资源实例的标签信息
- 增强型跨源连接相关API
  - 创建增强型跨源连接
  - 删除增强型跨源连接
  - 查询增强型跨源连接列表
  - 查询增强型跨源连接
  - 绑定队列
  - 解绑队列
  - 修改主机信息
  - 查询增强型跨源授权信息
  - 创建路由
  - 删除路由
- 跨源认证相关API
  - 创建跨源认证
  - 获取跨源认证列表
  - 更新跨源认证
  - 删除跨源认证
- 弹性资源池相关API
  - 创建弹性资源池
  - 查询所有弹性资源池
  - 删除弹性资源池
  - 修改弹性资源池信息
  - 查询弹性资源池下所有队列
  - 关联队列到弹性资源池
  - 弹性资源池扩缩容历史记录
  - 修改弹性资源池关联队列的扩缩容策略
- 队列相关API（推荐）
  - 创建队列
  - 删除队列
  - 查询所有队列
  - 查询队列详情
  - 重启/扩容/缩容队列
  - 创建指定地址连通性测试请求
  - 查询指定地址连通性测试详情
- SQL作业相关API
  - 提交SQL作业（推荐）
  - 取消作业（推荐）
  - 查询所有作业
  - 预览SQL作业查询结果
  - 导出查询结果
  - 查询作业状态
  - 查询作业详细信息
  - 检查SQL语法
  - 查询作业执行进度信息
- SQL模板相关API
  - 保存SQL模板
  - 查看所有SQL模板
  - 更新SQL模板
  - 删除SQL模板
- Flink作业相关API
  - 新建SQL作业
  - 更新SQL作业
  - 新建Flink Jar作业
  - 更新Flink Jar作业
  - 批量运行作业
  - 查询作业列表
  - 查询作业详情
  - 查询作业执行计划
  - 批量停止作业
  - 删除作业
  - 批量删除作业
  - 导出Flink作业
  - 导入Flink作业
  - 生成Flink SQL作业的静态流图
- Flink作业模板相关API
  - 新建模板
  - 更新模板
  - 删除模板
  - 查询模板列表
- Flink作业管理相关API
  - 触发Flink作业保存点
  - 导入Flink作业保存点
- Spark作业相关API
  - 创建批处理作业
  - 查询批处理作业列表
  - 查询批处理作业详情
  - 查询批处理作业状态
  - 取消批处理作业
- Spark作业模板相关API
  - 创建作业模板
  - 查询作业模板列表
  - 修改作业模板
  - 获取作业模板
- 权限策略和授权项
- 历史API
  - 委托相关API（废弃）
    - 获取DLI委托信息（废弃）
    - 创建DLI委托（废弃）
  - 分组资源相关API（废弃）
  - Spark批处理相关API（废弃）
    - 查询批处理作业日志（废弃）
  - SQL作业相关API（废弃）
    - 导入数据（废弃）
    - 导出数据（废弃）
  - 资源相关API（废弃）
    - 数据库相关API（废弃）
    - 表相关API（废弃）
  - 权限相关API（废弃）
  - 队列相关API（废弃）
  - 跨源认证相关API（废弃）
  - 增强型跨源连接相关API（废弃）
    - 创建路由（废弃）
    - 删除路由（废弃）
  - 模板相关API（废弃）
    - 查询所有SQL样例模板（废弃）
  - 表相关API（废弃）
    - 查询所有表（废弃）
  - SQL作业相关API（废弃）
  - 上传数据相关API（废弃）
    - 对已创建的上传作业进行鉴权（废弃）
  - 集群相关API（废弃）
  - Flink作业相关API（废弃）
    - 查询作业监控信息（废弃）
    - OBS授权给DLI服务
- 公共参数
  - 状态码
  - 错误码
  - 获取项目ID
  - 获取账号ID
SDK参考
- DLI SDK简介
- （推荐）DLI SDK V3
- DLI SDK
  - DLI SDK功能矩阵
  - DLI SDK与API的对应关系
- Java SDK
- Python SDK
常见问题
- DLI产品咨询类
- DLI弹性资源池和队列类
- DLI数据库和表类
- 增强型跨源连接类
- SQL作业类
  - SQL作业开发类
  - SQL作业运维类
- Flink作业类
- Spark作业相类
  - Spark作业开发类
  - Spark作业运维类
- DLI资源配额类
- DLI权限管理类
- DLI API类
视频帮助
更多文档
- 用户指南（阿布扎比区域）
- API参考（阿布扎比区域）
- SQL语法参考（阿布扎比区域）
- 用户指南（巴黎区域）
- API参考 (巴黎区域)
- SQL语法参考（巴黎区域）
- 用户指南（吉隆坡区域）
- API参考（吉隆坡区域）
- SQL语法参考（吉隆坡区域）
通用参考
- 云服务等级协议（SLA）
- 白皮书资源
- 支持区域
- 系统权限

本文导读

操作场景
操作流程
准备工作
步骤1：准备数据源通道
步骤2：准备数据输出通道
步骤3：创建OBS桶保存输出数据
步骤4：创建弹性资源池并添加队列
步骤5：创建DLI连接Kafka的增强型跨源连接
步骤6：创建DLI连接RDS的增强型跨源连接
步骤7：使用DEW管理访问凭据，并配置允许DLI访问DEW的委托
步骤8：创建Flink OpenSource SQL作业

展开导读

文档首页/ 数据湖探索 DLI/ 快速入门/ 使用DLI提交Flink OpenSource SQL作业查询RDS MySQL数据

使用DLI提交Flink OpenSource SQL作业查询RDS MySQL数据

更新时间：2025-01-09 GMT+08:00

查看PDF

操作场景

DLI Flink作业支持使用其他服务作为数据源和数据输出通道进行数据实时计算操作。

本例采用Kafka服务作为数据源通道，以RDS作为数据输出通道，介绍创建并提交Flink OpenSource SQL作业进行实时计算的操作步骤。

操作流程

样例场景需要创建一个Flink OpenSource SQL作业，且该作业包含一个输入流和一个输出流。输入流用于从Kafka读取数据，输出流用于将数据写入到RDS中。操作流程如操作流程所示。

开始进行如下操作前，请务必参考准备工作完成必要操作。

表1 使用DLI提交SQL作业查询RDS MySQL数据的操作流程
操作步骤	说明
步骤1：准备数据源通道	本样例场景需要创建Kafka实例作为数据源通道。
步骤2：准备数据输出通道	本样例场景需要创建RDS实例作为数据输出通道。
步骤3：创建OBS桶保存输出数据	创建OBS桶，为DLI Flink作业提供Checkpoint、保存作业日志和调试测试数据的存储功能。
步骤4：创建弹性资源池并添加队列	创建提交Flink作业所需的计算资源。
步骤5：创建DLI连接Kafka的增强型跨源连接	通过增强型跨源连接建立DLI弹性资源池与Kafka实例的网络连通。
步骤6：创建DLI连接RDS的增强型跨源连接	通过增强型跨源连接建立DLI弹性资源池与RDS实例的网络连通。
步骤7：使用DEW管理访问凭据，并配置允许DLI访问DEW的委托	跨源分析场景中，使用DEW管理数据源的访问凭证，并创建允许DLI访问DEW的委托。
步骤8：创建Flink OpenSource SQL作业	准备好数据源和数据输出通道后创建Flink OpenSource SQL作业分析数据。

准备工作

已注册华为账号并开通华为云，且在使用DLI前检查账号状态，账号不能处于欠费或冻结状态。
配置DLI委托访问授权
DLI使用过程中涉及到OBS、VPC、SMN等服务交互，首次使用DLI需要用户配置委托授权，允许访问这些依赖服务。
1. 使用华为云账号登录DLI管理控制台，在左侧导航栏单击“全局配置 > 服务授权”。
2. 在委托设置页面，勾选基础使用、跨源场景、运维场景的委托权限后，单击“更新委托权限”。
3. 查看并了解更新委托的提示信息，单击“确定”。完成DLI委托权限的更新。
  图1 配置DLI委托访问授权
4. 完成配置后，在IAM控制台的委托列表中，可查看到dli_management_agency的委托信息。

步骤1：准备数据源通道

本例以Kafka数据作为数据源通道。

开通Kafka数据接入服务，具体操作请参见创建Kafka实例。

创建Kafka相关依赖资源
在创建Kafka实例前您需要提前准备相关依赖资源，包括VPC、子网和安全组，并配置安全组。
- 创建VPC和子网的操作指导请参考创建虚拟私有云和子网，如果需要在已有VPC上创建和使用新的子网，请参考为虚拟私有云创建新的子网。
  说明：
  创建的VPC与使用的Kafka服务应在相同的区域。
  
  创建VPC和子网时，如无特殊需求，配置参数使用默认配置即可。
- 创建安全组的操作指导请参考创建安全组，为安全组添加规则的操作指导请参考添加安全组规则。
更多信息请参考《分布式消息服务Kafka用户指南》中的“准备实例依赖资源”章节。

创建用于作业输入流的Kafka专享版实例

登录分布式消息服务Kafka管理控制台。
在管理控制台左上角选择区域。

在“Kafka专享版”页面，单击右上角“购买Kafka实例”配置相关参数。实例信息如下：

更多Kafka专享版实例参数配置说明请参考购买Kafka实例。

表2 购买Kafka专享版实例参数说明
参数名称	参数说明	配置样例
区域	不同区域的云服务产品之间内网互不相通。请就近选择靠近您业务的区域，可减少网络时延，提高访问速度。	华东-上海二
可用区	可用区指在同一区域下，电力、网络隔离的物理区域，可用区之间内网互通，不同可用区之间物理隔离。	可用区1、可用区2、可用区3
版本	Kafka的版本号，实例创建后，版本号不支持修改。	3.x
部署架构	根据需求选择“单机”或“集群”。	集群
代理规格	请根据业务需求选择相应的代理规格。	kafka.2u4g.cluster.small
代理数量	选择代理数量。	3
单个代理存储空间	选择存储Kafka数据的磁盘类型和磁盘大小。Kafka实例创建后，磁盘类型不支持修改。	高I/O 100GB
虚拟私有云	选择已经创建好的或共享的虚拟私有云。	选择1中创建的虚拟私有云
子网	选择已经创建好的或共享的子网。子网在Kafka实例创建完成后，不支持修改。	选择1中创建的子网
安全组	选择已经创建好的安全组。	选择1中创建的安全组
接入方式	接入方式分为以下两种：明文接入：表示客户端连接Kafka实例时，无需进行SASL认证。密文接入：表示客户端连接Kafka实例时，需要进行SASL认证。	明文接入
实例名称	按命名规则自定义实例名称。	kafka-dliflink
企业项目	企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理，默认项目为default。	default

单击“立即购买”，弹出规格确认页面。
确认实例信息无误且阅读并同意《华为云用户协议》后，单击“提交”，完成实例创建,创建实例大约需要10~15分钟请耐心等待。

创建Kafka的topic。
1. 单击购买的Kafka实例名称，进入到Kafka实例的基本信息页面。
2. 单击“Topic管理 > 创建Topic”，创建一个Topic。Topic配置参数如下：
  - Topic名称。本示例输入为：testkafkatopic。
  - 分区数：1。
  - 副本数：1。
  其他参数保持默认即可。

步骤2：准备数据输出通道

采用RDS作为数据输出通道，创建RDS MySQL实例。

登录RDS管理控制台。
在管理控制台左上角选择区域。

单击“购买数据库实例”配置相关参数。主要参数的填下说明如下，其他参数保持默认值即可。

详细参数说明请参考购买RDS for MySQL实例。

表3 RDS MySQL实例参数配置信息
参数名称	参数说明	取值样例
计费模式	选择RDS实例的计费模式。	按需计费
区域	选择与DLI服务相同的区域。	华东-上海二
实例名称	实例名称。	rds-dliflink
数据库引擎	MySQL	MySQL
数据库版本	选用RDS for MySQL数据库时，请根据实际业务需求选择合适的数据库引擎版本。建议您选择当前可用的最高版本数据库，因其性能更稳定，安全性更高，使用更可靠。	8.0
实例类型	选择实例的主备类型。	主备
存储类型	实例的存储类型决定实例的读写速度。最大吞吐量越高，读写速度越快。	SSD云盘
可用区	对于单机实例，仅需选择单个可用区。	自定义
时区	由于世界各国家与地区经度不同，地方时也有所不同，因此会划分为不同的时区。时区可在创建实例时选择，后期可修改。	默认
性能规格	实例的CPU和内存。不同性能规格对应不同连接数和最大IOPS。	2vCPUs \| 4GB
存储空间	如果存储类型为SSD云盘或极速型SSD，可设置存储空间自动扩容，当存储空间可用率过小时，会自动扩容存储空间。	40GB
磁盘加密	选择是否开启磁盘加密功能。	不加密
虚拟私有云、子网	选择已有的虚拟私有云和子网。如需重新创建VPC和子网，请参考。说明：跨源场景数据源网段和弹性资源池的网段不能重合。	选择1中创建的虚拟私有云和子网。
数据库端口	默认使用3306端口。	3306
安全组	安全组限制实例的安全访问规则，加强云数据库RDS服务与其他服务间的安全访问。数据源的安全组需放通DLI弹性资源池的网段。	选择1中创建的安全组。
设置密码	设置实例的登录密码。	-
管理员账号	root	root
管理员密码	设置管理员密码。	-
参数模板	数据库参数模板就像是数据库引擎配置值的容器，参数模板中的参数可应用于一个或多个相同类型的数据库实例。	Default-MySQL-8.0
企业项目	对于已成功关联企业项目的用户，仅需在“企业项目”下拉框中选择目标项目。	default
购买数量	实例购买数量。	1

单击“立即购买”，确认规格信息。
单击“提交”，完成RDS实例的创建。

登录MySQL，并使用下述命令在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;

步骤3：创建OBS桶保存输出数据

在本样例中，需要为作业“JobSample”开通对象存储服务（OBS），为DLI Flink作业提供Checkpoint、保存作业日志和调试测试数据的存储功能。

具体操作请参见《对象存储服务控制台指南》中的“创建桶”章节。

在OBS管理控制台左侧导航栏选择“对象存储”。

在页面右上角单击“创建桶”，配置桶参数。

表4 OBS桶参数说明
参数名称	参数说明	取值样例
区域	桶所属区域。请选择与DLI服务相同的区域。	华东-上海二
桶名称	桶的名称。需全局唯一，不能与已有的任何桶名称重复，包括其他用户创建的桶。桶创建成功后，不支持修改名称，创建时，请设置合适的桶名。	obstest
默认存储类别	桶的存储类别。不同的存储类别可以满足客户业务对存储性能、成本的不同诉求。	标准存储
桶策略	桶的读写权限控制。私有桶除桶ACL授权外的其他用户无桶的访问权限。	私有
企业项目	将桶加入到企业项目中统一管理。	default

单击“立即创建”。

步骤4：创建弹性资源池并添加队列

创建DLI Flink OpenSource SQL作业，不能使用系统已有的default队列，需要您创建新队列，本例创建弹性资源池“dli_resource_pool”、队列“dli_queue_01”。

登录DLI管理控制台。
在左侧导航栏单击“资源管理 > 弹性资源池”，可进入弹性资源池管理页面。
在弹性资源池管理界面，单击界面右上角的“购买弹性资源池”。

在“购买弹性资源池”界面，填写具体的弹性资源池参数。

本例在华东-上海二区域购买按需计费的弹性资源池。相关参数说明如表5所示。

表5 参数说明
参数名称	参数说明	配置样例
区域	选择弹性资源池所在区域。	华东-上海二
项目	每个区域默认对应一个项目，由系统预置。	系统默认项目
名称	弹性资源池名称。	dli_resource_pool
规格	选择弹性资源池规格。	标准版
CU范围	弹性资源池最大最小CU范围。	64-64
网段	规划弹性资源池所属的网段。如需使用DLI增强型跨源，弹性资源池网段与数据源网段不能重合。弹性资源池网段设置后不支持更改。	172.16.0.0/19
企业项目	选择对应的企业项目。	default

参数填写完成后，单击“立即购买”，在界面上确认当前配置是否正确。
单击“提交”完成弹性资源池的创建。
在弹性资源池的列表页，选择要操作的弹性资源池，单击操作列的“添加队列”。

配置队列的基础配置，具体参数信息如下。

表6 弹性资源池添加队列基础配置
参数名称	参数说明	配置样例
名称	弹性资源池添加的队列名称。	dli_queue_01
类型	选择创建的队列类型。执行SQL作业请选择SQL队列。执行Flink或Spark作业请选择通用队列。	_
执行引擎	SQL队列可以选择队列引擎为Spark或者HetuEngine。	_
企业项目	选择对应的企业项目。	default

单击“下一步”，配置队列的扩缩容策略。

单击“新增”，可以添加不同优先级、时间段、“最小CU”和“最大CU”扩缩容策略。

本例配置的扩缩容策略如图2所示。

图2 添加队列时配置扩缩容策略
点击放大

表7 扩缩容策略参数说明
参数名称	参数说明	配置样例
优先级	当前弹性资源池中的优先级数字越大表示优先级越高。本例设置一条扩缩容策略，默认优先级为1。	1
时间段	首条扩缩容策略是默认策略，不能删除和修改时间段配置。即设置00-24点的扩缩容策略。	00-24
最小CU	设置扩缩容策略支持的最小CU数。	16
最大CU	当前扩缩容策略支持的最大CU数。	64

单击“确定”完成添加队列配置。

步骤5：创建DLI连接Kafka的增强型跨源连接

创建DLI Flink作业，还需要创建增强型跨源连接。具体操作请参考《数据湖探索用户指南》>《跨源连接》>《增强型跨源连接》。

说明：

增强型跨源仅支持按需专属队列。
绑定跨源的DLI队列网段和数据源网段不能重合。
系统default队列不支持创建跨源连接。
访问跨源表需要使用已经创建跨源连接的队列。

在Kafka的安全组上放通DLI队列网段
1. 在Kafka管理控制台，选择“Kafka专享版”，单击对应的Kafka名称，进入到Kafka的基本信息页面。
2. 在“连接信息”中获取该Kafka的“内网连接地址”，在“基本信息”的“网络”中获取该实例的“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
3. 单击“网络”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。
  例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
创建DLI队列连接Kafka的增强型跨源连接。
1. 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
2. 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
  - 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_kafka。
  - 弹性资源池：选择步骤4：创建弹性资源池并添加队列中已经创建的资源名称。
  - 虚拟私有云：选择Kafka的虚拟私有云。
  - 子网：选择Kafka的子网。
  - 其他参数可以根据需要选择配置。
  参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为“已激活”后可以进行后续步骤。
3. 单击“资源管理 > 队列管理”，选择操作的队列，本示例为步骤4：创建弹性资源池并添加队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
4. 在“测试连通性”界面，地址栏输入“Kafka内网地址:Kafka数据库端口”，单击“测试”测试DLI到Kafka网络是否可达。注意多个地址要分开单独测试。

步骤6：创建DLI连接RDS的增强型跨源连接

在RDS的安全组上放通DLI队列网段。
如果RDS和Kafka在同一VPC下的同一安全组，则无需重复执行此步骤，在步骤1中已在该安全组放通DLI的队列网段。
1. 在RDS管理控制台，选择“实例管理”，单击对应的RDS实例名称，进入到RDS的基本信息页面。
2. 在“基本信息”的“连接信息”中获取该实例的“内网地址”、“数据库端口”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
3. 单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
创建DLI队列连接RDS的增强型跨源连接。

如果RDS和Kafka在同一VPC和子网，则无需重复执行此步骤，在步骤2中创建的增强型跨源连接已将网络打通。
如果Kafka和RDS实例分别在两个VPC和子网下，则要执行以下步骤创建DLI队列连接RDS的增强型跨源连接。
1. 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
2. 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
  - 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_rds。
  - 弹性资源池：选择步骤4：创建弹性资源池并添加队列中已经创建的队列名称。
  - 虚拟私有云：选择RDS的虚拟私有云。
  - 子网：选择RDS的子网。
  - 其他参数可以根据需要选择配置。
  参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。
3. 单击“资源管理 > 队列管理”，选择操作的队列，本示例为步骤4：创建弹性资源池并添加队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
4. 在“测试连通性”界面，地址栏输入“RDS内网地址:RDS数据库端口”，单击“测试”测试DLI到RDS网络是否可达。

步骤7：使用DEW管理访问凭据，并配置允许DLI访问DEW的委托

跨源分析场景中，需要在connector中设置账号、密码等属性。但是账号密码等信息属于高度敏感数据，需要做加密处理，以保障用户的数据隐私安全。Flink1.15版本支持使用DEW管理凭据，在执行作业前需创建自定义委托并在作业中配置委托信息。

数据加密服务（Data Encryption Workshop，DEW）、云凭据管理服务（Cloud Secret Management Service，CSMS），提供一种安全、可靠、简单易用隐私数据加解密方案。本例介绍Flink Opensource SQL使用DEW管理RDS访问凭据的配置方法。

创建DLI访问DEW的委托并完成委托授权。
在DEW创建通用凭证。
登录DEW管理控制台
选择“凭据管理”，进入“凭据管理”页面。
单击“创建凭据”，配置凭据基本信息
- 凭据名称：待创建凭据的名称。本例名称为secretInfo。
- 凭据值：配置RDS实例的用户名和密码。
  - 第一行凭据值的键为MySQLUsername，值为RDS实例的用户名。
  - 第二行凭据值的键为MySQLPassword，值为RDS实例的密码。
  图3 设置凭据值
按需完成其他参数的配置后，单击“确定”保存凭据。

步骤8：创建Flink OpenSource SQL作业

准备好数据源和数据输出通道之后，就可以创建Flink OpenSource SQL作业了。

在DLI管理控制台的左侧导航栏中，单击“作业管理 > Flink作业”，进入“Flink作业”页面。
在“Flink作业”页面右上角单击“创建作业”，弹出“创建作业”对话框。配置参数：
- 类型：Flink OpenSource SQL
- 名称：JobSample
- 描述：不填
- 模板名称：不选择
- 标签：不填
单击“确定”，进入作业“编辑”页面。
设置作业运行参数。配置必选参数：
- 所属队列：dli_queue_01
- Flink版本：Flink 1.12。
- 保存作业日志：勾选。
- OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。
- 开启Checkpoint：勾选。
其余参数可不用配置。
单击“保存”，保存作业和相关参数。

编辑Flink OpenSource SQL作业。

在Flink OpenSource SQL语句编辑区域，输入详细的Flink OpenSource 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.0.237:9092,192.168.0.252:9092,192.168.0.137:9092", // 替换为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
) WITH (
  'connector' = 'jdbc',
  'url' = "jdbc:mysql://172.16.0.116:3306/rds-dliflink",// testrdsdb为创建的RDS的数据库名，IP和端口替换为RDS MySQL的实例IP和端口
  'table-name' = 'orders',
  'pwd_auth_name'="xxxxx", // DLI侧创建的Password类型的跨源认证名称，使用跨源认证则无需在作业中配置账号和密码
  'sink.buffer-flush.max-rows' = '1'
);

insert into jdbcSink select * from kafkaSource;

单击“语义校验”，确保语义校验成功。
单击“启动”，进入“启动Flink作业”页面，确认作业规格和费用后，单击“立即启动”，启动作业。
启动作业后，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在“状态”列中可以查看作业状态。作业提交成功后，状态将由“提交中”变为“运行中”。

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

连接Kafka集群，向Kafka相应的topic中发送如下测试数据：

Kafka生产和发送数据的方法请参考：连接实例生产消费信息。

{"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} 

{"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}

查看表中数据，在MySQL中执行sql查询语句。

select * from orders；

其结果参考如下（以下数据为从MySQL中复制的结果）。

202103241000000001,webShop,2021-03-24 10:00:00,100.0,100.0,2021-03-24 10:02:03,0001,Alice,330106
202103241606060001,appShop,2021-03-24 16:06:06,200.0,180.0,2021-03-24 16:10:06,0001,Alice,330106

上一篇：使用DLI提交SQL作业查询RDS MySQL数据

下一篇：使用DLI提交Flink Jar作业

意见反馈

文档内容是否对您有帮助？

有帮助没帮助

提供反馈

提交成功！非常感谢您的反馈，我们会继续努力做到更好！

系统繁忙，请稍后重试

在使用文档中是否遇到以下问题

内容与产品页面不一致

内容不易理解

缺失示例代码

步骤不可操作

搜不到想要的内容

缺少最佳实践

意见反馈（选填）

0/500

请至少选择一项反馈信息并填写问题反馈

字符长度不能超过500

直接提交取消