Flink mysql-cdc

Flink mysql-cdc

使用本地mysql 测试 mysql-cdc 数据到flink的过程：

一、mysql 配置

启动本地mysql

mysql开启binlog:

我的mysql安装地址：D:softstallmysql stall，  找打文件：my.ini

增加配置：

# Binary Logging.
#log-bin=D:softstallmysqlbinlog
log-bin=mysql-bin
binlog-format=Row
server-id=1001   # high version mysql

本地mysql 开启端口供外网访问：

Windows mysql 对外开放端口号:
https://www.cnblogs.com/zhurong/p/9398602.html

 

 二、配置作业sql 文件

CREATE TABLE `flink_test_order` (
  `id` STRING,
  `relate_id` STRING,
  `order_num` INT,
  `create1` TIMESTAMP,
  `create2` TIMESTAMP
)
with(
'hostname'='localhost',
'password'='******'
'connector'='mysql-cdc',
'port'='3306',
'database-name'='test001',
'table-name'='test_order',
'username'='root'
)
 
 CREATE TABLE `flink_test_order2` (
  `id` STRING,
  `relate_id` STRING,
  `order_num` INT,
  `create1` TIMESTAMP,
  `create2` TIMESTAMP
)
with('connector' = 'print')
 
 insert into
  flink_test_order2
select
  *
from
  flink_test_order

结果：

 

SELECt * FROM test001.test_order;

INSERT INTO test001.test_order
(id, relate_id, order_num, create1, create2)
VALUES('3', '3', 4, now(), now() );