Flink Opensource SQL使用DEW管理访问凭据

操作场景

DLI将Flink作业的输出数据写入到MySQL或DWS时，需要在Connector中设置账号、密码等敏感参数。但是这些信息如果使用明文形式存储存在数据安全风险，推荐做加密处理，以保障用户的数据隐私安全。

数据加密服务（Data Encryption Workshop，DEW）和云凭据管理服务（Cloud Secret Management Service，CSMS），提供一种安全、可靠、简单易用隐私数据加解密方案。只需将数据库账号和密码等信息托管为凭据，在Flink作业中配置引用该凭据即可通过安全通道获取所需的账号和密码信息。

且CSMS支持对凭据的全生命周期的统一管理，提升凭据管理的安全性和效率，有效避免程序硬编码或明文配置等问题导致的敏感信息泄露以及权限失控带来的业务风险。

本节操作介绍Flink Opensource SQL场景使用DEW管理和访问凭据的操作指导。

约束与限制

仅支持Flink1.15版本使用DEW管理访问凭据，在创建作业时，请配置作业使用Flink1.15版本、且已在作业中配置允许DLI访问DEW的委托信息。

前提条件

• 已在DEW服务创建通用凭证，并存入凭据值。具体操作请参考：创建通用凭据。
• 已创建DLI访问DEW的委托并完成委托授权。该委托需具备以下权限：
  • DEW中的查询凭据的版本与凭据值ShowSecretVersion接口权限，csms:secretVersion:get。
  • DEW中的查询凭据的版本列表ListSecretVersions接口权限，csms:secretVersion:list。
  • DEW解密凭据的权限，kms:dek:decrypt。

  委托权限示例请参考自定义DLI委托权限和常见场景的委托权限策略。

• 在DLI管理控制台新建“增强型跨源连接”配置DLI与数据源的网络连通。

  具体操作请参考增强型跨源连接。

语法格式

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'=''
  
);

参数说明

示例

本例以通过DataGen表产生随机数据并输出到MySQL结果表中为例，介绍Flink Opensource SQL使用DEW管理访问凭据的配置方法。

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

      了解更多请参考创建通用凭据。

2. 在Flink Opensource SQL作业中配置使用DEW管理访问凭据。

   请确保已创建DLI与MySQL的增强型跨源连接。详细步骤请参考创建增强型跨源连接。

   请确保已创建DLI访问DEW的委托并完成委托授权。详细步骤请参考自定义DLI委托权限。

   Flink Opensource 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;