双流实时同步

本文提供Flink双流实时同步场景的实施思路、配置示例和验证方式，可作为实际项目配置参考。

案例

将公司表作为维表，关联订单表，完善订单信息后，过滤掉阿里云的订单，结果同步到订单详情表中。

-- 公司表
create table source_business
(
    id   varchar(100) not null primary key,
    name varchar(100) not null
);

-- 订单表
create table source_order
(
    id          varchar(100) not null primary key,
    name        varchar(100) not null,
    business_id varchar(100) not null,
    price       double       not null
);

-- 订单详情表
create table target_business_order
(
    order_id      varchar(100) not null primary key,
    name          varchar(100) not null,
    business_name varchar(100) not null,
    price         double       not null
);

维表关联写法(Lookup Join)

使用Flink SQL语法 FOR SYSTEM_TIME AS OF

CREATE TABLE source_business
(
    id STRING,
    name STRING
) WITH ( 
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:30306/isxcode_db?useSSL=false&allowPublicKeyRetrieval=true',
    'driver' = 'com.mysql.cj.jdbc.Driver',
    'table-name' = 'source_business',
    'username' = 'root',
    'password' = 'root123'
);

CREATE TABLE source_order
(
    id STRING,
    name STRING,
    business_id STRING,
    price DOUBLE,
    order_time as proctime()
) WITH ( 
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '30306',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'isxcode_db',
    'table-name' = 'source_order',
    'scan.incremental.snapshot.enabled' = 'false',
    'server-time-zone' = 'UTC',
    'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE target_business_order
(
    order_id STRING,
    name STRING,
    business_name STRING,
    price DOUBLE,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH ( 
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:30306/isxcode_db?useSSL=false&allowPublicKeyRetrieval=true',
    'driver' = 'com.mysql.cj.jdbc.Driver',
    'table-name' = 'target_business_order',
    'username' = 'root',
    'password' = 'root123'
);

INSERT INTO target_business_order
SELECT
  o.id AS order_id,
  o.name,
  b.name,
  o.price
FROM source_order as o
LEFT JOIN source_business FOR SYSTEM_TIME AS OF o.order_time as b
  ON o.business_id = b.id where b.name <> '阿里云';

普通Join写法

FULL OUTER JOIN

外部链接，插入任意一条数据都会缓存 前提，每张表中至少存在一条数据，否则数据不合法，无法缓存 比如：source_business表为空，source_order表插入再多的数据也都是不合法的 当source_business插入一条数据后，source_order再插入的数据都会合法

CREATE TABLE source_business
(
    id STRING,
    name STRING
) WITH ( 
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '30306',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'isxcode_db',
    'table-name' = 'source_business',
    'scan.incremental.snapshot.enabled' = 'false',
    'server-time-zone' = 'UTC',
    'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE source_order
(
    id STRING,
    name STRING,
    business_id STRING,
    price DOUBLE
) WITH ( 
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '30306',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'isxcode_db',
    'table-name' = 'source_order',
    'scan.incremental.snapshot.enabled' = 'false',
    'server-time-zone' = 'UTC',
    'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE target_business_order
(
    order_id STRING,
    name STRING,
    business_name STRING,
    price DOUBLE
) WITH ( 
    'connector' = 'print' 
);

INSERT INTO target_business_order(order_id, name, business_name, price)
select o.id, o.name, b.name, o.price
from source_order o FULL OUTER join source_business b on o.business_id = b.id;

INNER JOIN

内链接，当数据同时插入，数据匹配上才会合法，并缓存数据，否则数据无法缓存

INSERT INTO target_business_order(order_id, name, business_name, price)
select o.id, o.name, b.name, o.price
from source_order o  INNER JOIN source_business b on o.business_id = b.id;

RIGHT JOIN

右链接，source_business有几条缓存几条，source_order匹配几条缓存几条

INSERT INTO target_business_order(order_id, name, business_name, price)
select o.id, o.name, b.name, o.price
from source_order o RIGHT JOIN source_business b on o.business_id = b.id;

LEFT JOIN

左链接，source_order有几条缓存几条，source_business匹配几条缓存几条

INSERT INTO target_business_order(order_id, name, business_name, price)
select o.id, o.name, b.name, o.price
from source_order o LEFT JOIN source_business b on o.business_id = b.id;

时间间隔Join写法

间隔链接，一段时间内的数据匹配

CREATE TABLE source_business
(
    id STRING,
    name STRING,
    rtime AS CURRENT_TIMESTAMP
) WITH ( 
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '30306',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'isxcode_db',
    'table-name' = 'source_business',
    'scan.incremental.snapshot.enabled' = 'false',
    'server-time-zone' = 'UTC',
    'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE source_order
(
    id STRING,
    name STRING,
    business_id STRING,
    price DOUBLE,
    order_time AS CURRENT_TIMESTAMP
) WITH ( 
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '30306',
    'username' = 'root',
    'password' = 'root123',
    'database-name' = 'isxcode_db',
    'table-name' = 'source_order',
    'scan.incremental.snapshot.enabled' = 'false',
    'server-time-zone' = 'UTC',
    'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE target_business_order
(
    order_id STRING,
    name STRING,
    business_name STRING,
    price DOUBLE
) WITH ( 
    'connector' = 'print' 
);

INSERT INTO target_business_order(order_id, name, business_name, price)
select o.id, o.name, b.name, o.price
from source_order o,
     source_business b
where o.business_id = b.id
  AND o.order_time BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '10' SECOND