Flink VVP+DLF数据入湖与分析实践

登录准备好的MySQL实例，详情请参见通过DMS登录RDS MySQL。

执行如下命令，创建一张表，并插入若干测试数据。

CREATE DATABASE testdb;
CREATE TABLE testdb.student (
  `id` bigint(20) NOT NULL,
  `name` varchar(256) DEFAULT NULL,
  `age` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`id`)
);

INSERT INTO testdb.student VALUES (1,'name1',10);
INSERT INTO testdb.student VALUES (2,'name2',20);

登录实时计算管理控制台。

进入创建Catalog页面。

1. 在Flink全托管页签，单击目标工作空间操作列下的控制台。

2. 在左侧导航栏，单击元数据管理。

3. 单击创建Catalog。

创建DLF Catalog。

1. 在创建Catalog页面，选择DLF，单击下一步。

2. 填写以下参数配置信息，单击确定。详情请参见管理DLF Catalog。

当您成功创建DLF之后，可在元数据管理中看到新增的dlf数据目录，默认链接的是DLF的default数据目录。

登录实时计算管理控制台。

在Flink全托管页签，单击目标工作空间操作列下的控制台。

创建数据源表和目标表。

1. 在左侧导航栏，单击数据开发 > 数据查询。

2. 在SQL编辑区域，输入以下代码，单击运行。

   -- 创建数据源表
   CREATE TABLE IF NOT EXISTS student_source (
     id INT,
     name VARCHAR (256),
     age INT,
     PRIMARY KEY (id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mysql-cdc',
     -- hostname替换为RDS的连接地址
     'hostname' = 'rm-xxxxxxxx.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = '<RDS user name>',
     'password' = '<RDS password>',
     'database-name' = '<RDS database>',
     -- table-name为数据源表，本例中填步骤二创建的student表
     'table-name' = 'student'
   );
   
   -- catalog名为步骤二创建的dlf catalog name，本例中填dlf
   CREATE DATABASE IF NOT EXISTS dlf.dlf_testdb;
   
   -- 创建目标表
   CREATE TABLE IF NOT EXISTS dlf.dlf_testdb.student_hudi (
     id    BIGINT PRIMARY KEY NOT ENFORCED,
     name  STRING,
     age    BIGINT
   ) WITH(
     'connector' = 'hudi'
   );

   创建成功后，可在元数据管理中看到新增的数据源表和目标表。

   

创建Flink SQL入湖作业。

1. 在左侧导航栏，单击数据开发 > ETL。

2. 单击新建后，在新建作业草稿对话框，选择空白的流作业草稿，单击下一步。

3. 填写作业信息，单击创建。

4. 在SQL编辑区域，输入以下代码，创建一个Flink SQL作业。

   -- 创建流SQL作业
   INSERT INTO dlf.dlf_testdb.student_hudi SELECT * FROM student_source  /*+ OPTIONS('server-id'='123456') */;

   说明

   • 关于MySQL源表的参数设置和使用条件，请参见MySQL的CDC源表。

   • 关于Hudi结果表的参数设置，请参见Hudi结果表。

5. 在SQL编辑区域右上方，单击部署。在部署新版本对话框，可根据需要填写或选中相关内容，单击确定。

启动作业。

1. 在左侧导航栏，单击运维中心 > 作业运维。

2. 单击目标作业名称操作列中的启动。

   选择无状态启动后，单击启动。当您看到作业状态变为运行中，则代表作业运行正常。作业启动参数配置，详情请参见作业启动。

在左侧导航栏，单击数据开发 > 数据查询。

在SQL编辑区域，输入以下代码，单击运行。

SELECT * FROM dlf.dlf_testdb.student_hudi;

查询结果如下图所示，可以直接对Flink写入数据湖的数据进行查询和分析。