Flink SQL使用DEW管理访问凭据
操作场景
AI DataLake将Flink作业的输出数据写入到MySQL或DWS时，需要在Connector中设置账号、密码等敏感参数。但是这些信息如果使用明文形式存储存在数据安全风险，推荐做加密处理，以保障用户的数据隐私安全。
密码安全中心（Data Encryption Workshop，DEW）和云凭据管理服务（Cloud Secret Management Service，CSMS），提供一种安全、可靠、简单易用隐私数据加解密方案。只需将数据库账号和密码等信息托管为凭据，在Flink作业中配置引用该凭据即可通过安全通道获取所需的账号和密码信息。
且CSMS支持对凭据的全生命周期的统一管理，提升凭据管理的安全性和效率，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。
本节操作介绍Flink SQL场景使用DEW管理和访问凭据的操作指导。
前提条件
- 已在DEW服务创建通用凭据，并存入凭据值。具体操作请参考：创建通用凭据。
- 已创建AI DataLake访问DEW的委托并完成Flink端点委托授权。该委托需具备以下权限：
- DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。
- DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。
- DEW解密凭据的权限，kms:dek:decrypt。
委托权限示例请参考创建自定义委托和常见场景的委托权限策略。
语法格式
create table tableName(
attr_name attr_type
(',' attr_name attr_type)*
(',' WATERMARK FOR rowtime_column_name AS watermark-strategy_expression)
)
with (
...
'dew.endpoint'='',
'dew.csms.secretName'='',
'dew.csms.decrypt.fields'='',
'dew.projectId'='',
'dew.csms.version'=''
); 参数说明
| 参数 | 是否必选 | 默认值 | 数据类型 | 参数说明 |
|---|---|---|---|---|
| dew.endpoint | 是 | 无 | String | 指定要使用的DEW服务所在的Endpoint信息。 获取地区和终端节点。 配置示例：'dew.endpoint'='kms.cn-xxxx.myhuaweicloud.com' |
| dew.projectId | 否 | Flink作业所在的项目ID | String | DEW所在的项目ID。 |
| dew.csms.secretName | 是 | 无 | String | 在DEW服务的凭据管理中新建的通用凭据的名称。 配置示例：'dew.csms.secretName'='secretInfo' |
| dew.csms.decrypt.fields | 是 | 无 | String | 指定connector的with属性中，哪些字段属性需要使用DEW云凭据管理服务进行解密。 字段属性之间用逗号分隔，例如：'dew.csms.decrypt.fields'='field1,field2,field3' |
| dew.csms.version | 否 | 最新的version | String | 在DEW服务的凭据管理中新建的通用凭据的版本号(凭据的版本标识符)。
|
示例
本例以通过DataGen表产生随机数据并输出到MySQL结果表中为例，介绍Flink SQL使用DEW管理访问凭据的配置方法。
- 在DEW创建通用凭据。具体操作请参考：创建通用凭据。
- 凭据名称：待创建凭据的名称。本例名称为secretInfo。
- 凭据值：配置RDS实例的用户名和密码。
- 第一行凭据值的键为MySQLUsername，值为RDS实例的用户名。
- 第二行凭据值的键为MySQLPassword，值为RDS实例的密码。
图1 设置凭据值
- 在Flink SQL作业中配置使用DEW管理访问凭据。
请确保已创建AI DataLake访问DEW的委托并完成委托授权。详情请参见创建自定义委托和常见场景的委托权限策略。
Flink SQL作业配置示例如下：1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
create table dataGenSource( user_id string, amount int ) with ( 'connector' = 'datagen', 'rows-per-second' = '1', --每秒生成一条数据 'fields.user_id.kind' = 'random', --为字段user_id指定random生成器 'fields.user_id.length' = '3' --限制user_id长度为3个字符 ); CREATE TABLE jdbcSink ( user_id string, amount int ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--其中url中的flink表示MySQL中orders表所在的数据库名 'table-name' = 'orders', 'username' = 'MySQLUsername', -- DEW服务中，名称为secretInfo，且版本号v1的通用凭据，定义凭证值的键MySQLUsername，它的值为用户的敏感信息。 'password' = 'MySQLPassword', -- DEW服务中，名称为secretInfo，且版本号v1的通用凭据，定义凭证值的键MySQLPassword，它的值为用户的敏感信息。 'sink.buffer-flush.max-rows' = '1', 'dew.endpoint'='endpoint', --使用的DEW服务所在的endpoint信息 'dew.csms.secretName'='secretInfo', --DEW服务通用凭据的凭据名称 'dew.csms.decrypt.fields'='username,password', --其中username,password字段值，需要利用DEW凭证管理,进行解密替换。 'dew.csms.version'='v1' ); insert into jdbcSink select * from dataGenSource;
