文档首页/ 数据湖探索 DLI/ 开发指南/ Flink OpenSource SQL作业开发指南/ 从MySQL CDC源表读取数据写入到DWS
更新时间：2024-11-14 GMT+08:00
查看PDF
分享

从MySQL CDC源表读取数据写入到DWS

本指导仅适用于Flink 1.12版本。

场景描述

CDC是变更数据捕获（Change Data Capture）技术的缩写，它可以将源数据库的增量变动记录，同步到一个或多个数据目的中。CDC在数据同步过程中，还可以对数据进行一定的处理，例如分组（GROUP BY）、多表的关联（JOIN）等。

本示例通过创建MySQL CDC源表来监控MySQL的数据变化，并将变化的数据信息插入到DWS数据库中。

前提条件

  1. 已创建RDS MySQL实例，具体步骤可参考：RDS MySQL快速入门。本示例创建的RDS MySQL数据库版本选择为：8.0。
  2. 已创建DWS实例，具体创建DWS集群的操作可以参考创建DWS集群。

整体作业开发流程

整体作业开发流程参考图1。
图1 作业开发流程

步骤1：创建队列：创建DLI作业运行的队列。

步骤2：创建RDS MySQL数据库和表：创建RDS MySQL的数据库和表。

步骤3：创建DWS数据库和表：创建用于接收数据的DWS数据库和表。

步骤4：创建增强型跨源连接：DLI上创建连接RDS和DWS的跨源连接，打通网络。

步骤5：运行作业：DLI上创建和运行Flink OpenSource作业。

步骤6：发送数据和查询结果：RDS MySQL的表上插入数据，在DWS上查看运行结果。

步骤1：创建队列

  1. 登录DLI管理控制台，在左侧导航栏单击“资源管理 > 队列管理”，可进入队列管理页面。
  2. 在队列管理界面，单击界面右上角的“购买队列”。
  3. 在“购买队列”界面，填写具体的队列配置参数，具体参数填写参考如下。
    • 计费模式：选择“包年/包月”或“按需计费”。本示例选择“按需计费”。
    • 区域和项目：保持默认值即可。
    • 名称：填写具体的队列名称。

      新建的队列名称，名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。长度限制：1~128个字符。

      队列名称不区分大小写，系统会自动转换为小写。

    • 类型：队列类型选择“通用队列”。“按需计费”时需要勾选“专属资源模式”。
    • AZ策略、CPU架构、规格：保持默认即可。
    • 企业项目：当前选择为“default”。
    • 高级选项：选择“自定义”。
    • 网段：配置队列网段。例如，当前配置为10.0.0.0/16。

      队列的网段不能和DMS Kafka、RDS MySQL实例的子网网段有重合，否则后续创建跨源连接会失败。

    • 其他参数根据需要选择和配置。
  4. 参数配置完成后，单击“立即购买”，确认配置信息无误后，单击“提交”完成队列创建。

步骤2：创建RDS MySQL数据库和表

  1. 登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS MySQL实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。
  2. 输入实例登录的用户名和密码。单击“登录”，即可进入RDS MySQL数据库并进行管理。
  3. 在数据库实例界面，单击“新建数据库”，数据库名定义为：testrdsdb，字符集保持默认即可。
  4. 在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，创建RDS MySQL表。 
    CREATE TABLE mysqlcdc (
	`order_id` VARCHAR(64) NOT NULL,
	`order_channel` VARCHAR(32) NOT NULL,
	`order_time` VARCHAR(32),
	`pay_amount` DOUBLE,
	`real_pay` DOUBLE,
	`pay_time` VARCHAR(32),
	`user_id` VARCHAR(32),
	`user_name` VARCHAR(32),
	`area_id` VARCHAR(32)

)	ENGINE = InnoDB
	DEFAULT CHARACTER SET = utf8mb4;

步骤3：创建DWS数据库和表

  1. 连接已创建的DWS集群。

    请参考使用gsql命令行客户端连接DWS集群。

  2. 执行以下命令连接DWS集群的默认数据库“gaussdb”： 
    gsql -d gaussdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
    • gaussdb：DWS集群默认数据库。
    • DWS集群连接地址：请参见获取集群连接地址进行获取。如果通过公网地址连接，请指定为集群“公网访问地址”或“公网访问域名”，如果通过内网地址连接，请指定为集群“内网访问地址”或“内网访问域名”。如果通过弹性负载均衡连接，请指定为“弹性负载均衡地址”。
    • dbadmin：创建集群时设置的默认管理员用户名。
    • -W：默认管理员用户的密码。
  3. 在命令行窗口输入以下命令创建数据库“testdwsdb”。 
    CREATE DATABASE testdwsdb;
  4. 执行以下命令，退出gaussdb数据库，连接新创建的数据库“testdwsdb”。 
    \q
gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
  5. 执行以下命令创建表。 
    create schema test;
set current_schema= test;
drop table if exists dwsresult;
CREATE TABLE dwsresult
(
    car_id VARCHAR, 
    car_owner VARCHAR,
    car_age INTEGER ,
    average_speed FLOAT8,
    total_miles FLOAT8
);

步骤4：创建增强型跨源连接

  • 创建DLI连接RDS的增强型跨源连接
    1. 在RDS管理控制台，选择“实例管理”，单击对应的RDS实例名称，进入到RDS的基本信息页面。
    2. 在“基本信息”的“连接信息”中获取该实例的“内网地址”、“数据库端口”、“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
    3. 单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
    4. 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
    5. 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
      • 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_rds。
      • 弹性资源池：选择步骤1：创建队列中已经创建的队列。
      • 虚拟私有云：选择RDS的虚拟私有云。
      • 子网：选择RDS的子网。
      • 其他参数可以根据需要选择配置。

      参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。

    6. 单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
    7. 在“测试连通性”界面，根据2中获取的RDS连接信息，地址栏输入“RDS内网地址:RDS数据库端口”，单击“测试”测试DLI到RDS网络是否可达。
  • 创建DLI连接DWS的增强型跨源连接
    1. 在DWS管理控制台，选择“集群管理”，单击已创建的DWS集群名称，进入到DWS的基本信息页面。
    2. 在“基本信息”的“数据库属性”中获取该实例的“内网IP”、“端口”，“基本信息”页面的“网络”中获取“虚拟私有云”和“子网”信息，方便后续操作步骤使用。
    3. 单击“连接信息”中的安全组名称，在“入方向规则”中添加放通队列网段的规则。例如，本示例队列网段为“10.0.0.0/16”，则规则添加为：优先级选择：1，策略选择：允许，协议选择：TCP，端口值不填，类型：IPv4，源地址为：10.0.0.0/16，单击“确定”完成安全组规则添加。
    4. RDS和DWS实例属于同一VPC和子网下？
      1. 是，执行7。RDS和DWS实例在同一VPC和子网，不用再重复创建增强型跨源连接。
      2. 否，执行5。RDS和DWS实例分别在两个VPC和子网下，则要分别创建增强型跨源连接打通网络。
    5. 登录DLI管理控制台，在左侧导航栏单击“跨源管理”，在跨源管理界面，单击“增强型跨源”，单击“创建”。
    6. 在增强型跨源创建界面，配置具体的跨源连接参数。具体参考如下。
      • 连接名称：设置具体的增强型跨源名称。本示例输入为：dli_dws。
      • 弹性资源池：选择步骤1：创建队列中已经创建的队列。
      • 虚拟私有云：选择DWS的虚拟私有云。
      • 子网：选择DWS的子网。
      • 其他参数可以根据需要选择配置。

      参数配置完成后，单击“确定”完成增强型跨源配置。单击创建的跨源连接名称，查看跨源连接的连接状态，等待连接状态为：“已激活”后可以进行后续步骤。

    7. 单击“队列管理”，选择操作的队列，本示例为步骤1：创建队列中创建的队列，在操作列，单击“更多 > 测试地址连通性”。
    8. 在“测试连通性”界面，根据2中获取的DWS连接信息，地址栏输入“DWS内网IP:DWS端口”，单击“测试”测试DLI到DWS网络是否可达。

步骤5：运行作业

  1. 在DLI管理控制台，单击“作业管理 > Flink作业”，在Flink作业管理界面，单击“创建作业”。
  2. 在创建队列界面，类型选择“Flink OpenSource SQL”，名称填写为：FlinkCDCMySQLDWS。单击“确定”，跳转到Flink作业编辑界面。
  3. 在Flink OpenSource SQL作业编辑界面，配置如下参数，其他参数默认即可。
    • 所属队列：选择步骤1：创建队列中创建的队列。
    • Flink版本：选择1.12。
    • 保存作业日志：勾选。
    • OBS桶：选择保存作业日志的OBS桶，根据提示进行OBS桶权限授权。
    • 开启Checkpoint：勾选。
    • Flink作业编辑框中输入具体的作业SQL，本示例作业参考如下。SQL中加粗的参数需要根据实际情况修改。

      本示例使用的Flink版本为1.12，故Flink OpenSource SQL语法也是1.12。本示例数据源是Kafka，写入结果数据到Elasticsearch。

      请参考Flink OpenSource SQL 1.12创建MySQL CDC源表和Flink OpenSource SQL 1.12创建DWS结果表。

      create table mysqlCdcSource(
  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' = 'mysql-cdc',
  'hostname' = '192.168.12.148',--IP替换为RDS MySQL的实例IP
  'port' = '3306',--端口替换为RDS MySQL的实例端口
  'pwd_auth_name'= 'xxxxx', --DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置账号和密码。
  'database-name' = 'testrdsdb',--RDS MySQL实例的数据库名
  'table-name' = 'mysqlcdc'--RDS MySQL实例的数据库下的表名
);

create table dwsSink(
  order_channel string,
  pay_amount double,
  real_pay double,
  primary key(order_channel) not enforced
) with (
  'connector' = 'gaussdb',
  'driver' = 'com.huawei.gauss200.jdbc.Driver',
  'url' = 'jdbc:gaussdb://192.168.168.16:8000/testdwsdb', ---192.168.168.16:8000替换为DWS的内网IP和端口，testdwsdb为创建的DWS数据库名
  'table-name' = 'test\".\"dwsresult', ---test为创建的DWS表的schema，dwsresult为对应的DWS表名
  'pwd_auth_name'= 'xxxxx', --DLI侧创建的Password类型的跨源认证名称。使用跨源认证则无需在作业中配置账号和密码。 
  'write.mode' = 'insert'
);

insert into dwsSink select order_channel, sum(pay_amount),sum(real_pay) from mysqlCdcSource group by order_channel;
  4. 单击“语义校验”确保SQL语义校验成功。单击“保存”，保存作业。单击“启动”，启动作业，确认作业参数信息，单击“立即启动”开始执行作业。等待作业运行状态变为“运行中”。

步骤6：发送数据和查询结果

  1. 登录RDS管理控制台，在“实例管理”界面，选择已创建的RDS MySQL实例，选择操作列的“更多 > 登录”，进入数据管理服务实例登录界面。
  2. 输入实例登录的用户名和密码。单击“登录”，即可进入RDS MySQL数据库并进行管理。
  3. 在已创建的数据库的操作列，单击“SQL查询”，输入以下创建表语句，插入测试数据。 
    insert into mysqlcdc values
('202103241000000001','webShop','2021-03-24 10:00:00','100.00','100.00','2021-03-24 10:02:03','0001','Alice','330106'),
('202103241206060001','appShop','2021-03-24 12:06:06','200.00','180.00','2021-03-24 16:10:06','0002','Jason','330106'),
('202103241403000001','webShop','2021-03-24 14:03:00','300.00','100.00','2021-03-24 10:02:03','0003','Lily','330106'),
('202103241636060001','appShop','2021-03-24 16:36:06','200.00','150.00','2021-03-24 16:10:06','0001','Henry','330106');
  4. 参考使用gsql命令行客户端连接DWS集群连接已创建的DWS集群。
  5. 执行以下命令连接DWS集群的默认数据库“testdwsdb”： 
    gsql -d testdwsdb -h DWS集群连接地址 -U dbadmin -p 8000 -W password -r
  6. 执行以下命令，查询DWS的表数据。 
    select * from test.dwsresult;
    查询结果参考如下： 
    order_channel  pay_amount  real_pay 
appShop         400.0       330.0
webShop         400.0       200.0

相关文档