基于阿里云KMS实现Flink作业数据库密码的加密与动态解密

1. 创建RDS MySQL数据库和账号。

   为目标实例创建名称为school的数据库和具有对应数据库读写权限的普通账号（flink_rds_user）。具体操作请参见第一步：快捷创建RDS MySQL实例与配置数据库。

2. 准备RDS MySQL数据源。

   1. 在目标实例详情页面，单击上方的登录数据库。

   2. 在弹出的DMS页面中，填写创建的数据库账号名和密码，然后单击登录。

   3. 登录成功后，在左侧双击school数据库，切换数据库。

   4. 在SQL Console区域编写三张业务表的建表DDL以及插入的数据语句。

      CREATE TABLE `student` (
        id INT not null primary key,
        username VARCHAR(255),
        age BIGINT
      );
        
      INSERT INTO student VALUES
      (001, 'lily', 15),
      (002, 'leilei', 18),
      (003, 'xiaoming', 17),
      (004, 'huahua', 15);
      
      SELECT * FROM student;

3. 单击执行，单击直接执行。

在对密钥进行加密之前，我们需要确保尚未对密钥进行加密的JAR文件能够正常运行。

1. 本地开发Flink JAR作业。

   1. 在IDEA上创建一个新项目。

   2. 将下面代码拷贝到名称为JavaDemo的类文件和POM文件中，并根据实际业务修改配置项取值。

      JavaDemo

      POM

      package org.example;
      
      import com.ververica.cdc.connectors.mysql.source.MySqlSource;
      import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema;
      
      import org.apache.flink.api.common.eventtime.WatermarkStrategy;
      import org.apache.flink.streaming.api.datastream.DataStreamSource;
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.table.api.DataTypes;
      import org.apache.flink.table.data.RowData;
      import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
      import org.apache.flink.table.types.DataType;
      import org.apache.flink.table.types.logical.LogicalType;
      import org.apache.flink.table.types.logical.RowType;
      import org.apache.flink.table.types.utils.TypeConversions;
      
      
      public class JavaDemo {
      
          public static void main(String[] args) throws Exception {
              final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      
              // 构建反序列化器
              DataType dataType =
                      DataTypes.ROW(
                              DataTypes.FIELD("id", DataTypes.INT()),
                              DataTypes.FIELD("username", DataTypes.STRING()),
                              DataTypes.FIELD("age", DataTypes.INT()));
      
              LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
              InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
              RowDataDebeziumDeserializeSchema deserializer =
                      RowDataDebeziumDeserializeSchema.newBuilder()
                              .setPhysicalRowType((RowType) dataType.getLogicalType())
                              .setResultTypeInfo(typeInfo)
                              .build();
      
              // 数据源(com.ververica.cdc.connectors.mysql.source.MySqlSource)
              MySqlSource<RowData> mySqlSource =
                      MySqlSource.<RowData>builder()
                              .hostname("rm-bp****2ye09w72zjq.mysql.rds.aliyuncs.com")
                              .port(3306)
                              .databaseList("school") // 设置数据库
                              .tableList("school.student") // 设置数据表
                              .username("flink_rds_user")
                              .password("flink_rds_password@123")
                              // 将数据初始化为RowData结构
                              .deserializer(deserializer)
                              .build();
      
              // 将外部数据源集成到flink数据流程序
              // WatermarkStrategy.noWatermarks() 指没有使用水印策略
              DataStreamSource<RowData> mySQLSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
      
              // 输出到日志stdout
              mySQLSource.print();
      
              //运行
              env.execute("MySQL CDC Test");
          }
      }
      

      <?xml version="1.0" encoding="UTF-8"?>
      <project xmlns="http://maven.apache.org/POM/4.0.0"
               xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
               xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
      
        <groupId>com.aliyun</groupId>
        <artifactId>JavaDemo</artifactId>
        <version>1.0-SNAPSHOT</version>
        <name>Flink MySQL CDC Demo</name>
      
        <properties>
          <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
          <maven.compiler.source>1.8</maven.compiler.source>
          <maven.compiler.target>1.8</maven.compiler.target>
          <flink.version>1.17.1</flink.version>
          <flink-cdc.version>2.4.2</flink-cdc.version>
          <log4j.version>2.17.1</log4j.version>
        </properties>
      
        <dependencies>
          <!-- Flink 核心依赖 (打包时用 provided) -->
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
          </dependency>
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
          </dependency>
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
          </dependency>
      
          <!-- Flink MySQL CDC 连接器 -->
          <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-mysql</artifactId>
            <version>1.17-vvr-8.0.4-1</version>
            <!-- 本地运行时注释掉下一行 -->
            <!-- <scope>provided</scope> -->
          </dependency>
      
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>1.17.1</version> <!-- 与核心版本严格一致 -->
            <scope>provided</scope>
          </dependency>
      
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
          </dependency>
          <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.17.1</version> <!-- 必须与核心版本严格一致 -->
            <scope>provided</scope>
          </dependency>
      
      
          <!-- 日志依赖 -->
          <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.17.1</version>
            <scope>runtime</scope>
          </dependency>
          
        </dependencies>
      
        <build>
          <plugins>
            <!-- 编译器插件 -->
            <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-compiler-plugin</artifactId>
              <version>3.13.0</version> <!-- 修正版本 -->
              <configuration>
                <source>${maven.compiler.source}</source>
                <target>${maven.compiler.target}</target>
                <encoding>UTF-8</encoding>
              </configuration>
            </plugin>
      
            <!-- 打包 Fat JAR -->
            <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-shade-plugin</artifactId>
              <version>3.5.1</version>
              <executions>
                <execution>
                  <phase>package</phase>
                  <goals>
                    <goal>shade</goal>
                  </goals>
                  <configuration>
                    <artifactSet>
                      <excludes>
                        <exclude>org.apache.flink:force-shading</exclude>
                        <exclude>com.google.code.findbugs:jsr305</exclude>
                        <!-- 保留日志依赖 -->
                        <!-- <exclude>org.slf4j:*</exclude> -->
                        <!-- <exclude>org.apache.logging.log4j:*</exclude> -->
                      </excludes>
                    </artifactSet>
                    <filters>
                      <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                          <exclude>META-INF/*.SF</exclude>
                          <exclude>META-INF/*.DSA</exclude>
                          <exclude>META-INF/*.RSA</exclude>
                          <exclude>META-INF/MANIFEST.MF</exclude> <!-- 新增关键过滤 -->
                        </excludes>
                      </filter>
                    </filters>
                    <transformers>
                      <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                        <mainClass>org.example.JavaDemo</mainClass>
                      </transformer>
                      <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> <!-- 新增必要转换器 -->
                    </transformers>
                  </configuration>
      
                </execution>
              </executions>
            </plugin>
          </plugins>
        </build>
      </project>
      

   3. 修改JavaDemo代码中的RDS MySQL实例连接信息。

      配置项	说明	示例值
      hostname	RDS MySQL的endpoint。	rm-bp****2ye09w72zjq.mysql.rds.aliyuncs.com
      port	RDS MySQL实例的端口号。	3306
      databaseList	RDS MySQL数据库名称。	school
      tableList	RDS MySQL表名称。	school.student
      username	RDS MySQL的用户名。	flink_rds_user
      password	RDS MySQL的用户密码。	flink_rds_password@123

   4. 在IDEA项目中进行打包。

      在IDEA项目的target目录查看已经打包好的JavaDemo-1.0-SNAPSHOT.jar文件。

2. 在实时计算开发控制台上部署、启动并查看运行结果。

   1. 将JavaDemo-1.0-SNAPSHOT.jar文件上传到实时计算开发控制台的文件管理。

   2. 部署JAR作业。

      1. 在运维中心 > 作业运维页面，单击部署作业，选择JAR作业。

      2. 填写部署信息。

         参数	说明	示例
         部署模式	请选择部署为流模式。	流模式
         部署名称	填写对应的JAR作业名称。	javademo
         引擎版本	当前作业使用的Flink引擎版本。 建议使用带有推荐、稳定标签的版本，这些版本具有更高的可靠性和性能表现，详情请参见功能发布记录和引擎版本介绍。	vvr-8.0.11-flink-1.17
         JAR URI	选择步骤4中上传的JAR包。 说明 实时计算引擎VVR 8.0.6及以上版本仅支持访问开通Flink工作空间时绑定的Bucket，不支持访问其他Bucket。	JavaDemo-1.0-SNAPSHOT.jar
         Entry Point Class	程序的入口类。如果您的JAR包未指定主类，请在此处输入您的Endpoint Class类的标准路径。	-
         Entry Point Main Arguments	填写传入参数信息，在主方法中调用该参数。	-
         部署目标	在下拉列表中，选择目标资源队列或者Session集群（请勿生产环境中使用）。详情请参见管理资源队列和步骤一：创建Session集群。 重要 部署到Session集群的作业不支持显示监控告警、配置监控告警和开启自动调优功能。请勿将Session集群用于正式生产环境，Session集群可以作为开发测试环境。详情请参见作业调试。	default-queue

         更多配置参数详情请参见部署作业。

      3. 单击部署。

   3. 在作业运维页面，单击javademo作业旁的启动，选择无状态启动后，单击启动。

   4. 在作业日志页面，查看Flink作业结果。

      作业状态变为运行中后，在TaskManager中以.out结尾的日志文件中，搜索lily查看Flink计算结果。