Dual-stream real-time synchronization

This article provides implementation ideas, configuration examples and verification methods for Flink dual-stream real-time synchronization scenarios, which can be used as a reference for actual project configuration.

Case

Use the company table as a dimension table and associate it with the order table. After completing the order information, filter out Alibaba Cloud orders and synchronize the results to the order details table.

-- 公司表
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
);

Dimension table association writing method (Lookup Join)

Use Flink SQL syntax 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 <> '阿里云';

Ordinary Join writing method

FULL OUTER JOIN

External links, any piece of data inserted will be cached The premise is that there is at least one piece of data in each table, otherwise the data is illegal and cannot be cached. For example: the source_business table is empty, and no matter how much data is inserted into the source_order table, it is illegal. When source_business inserts a piece of data, the data inserted by source_order will be legal.

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

Internal linking, when data is inserted at the same time, the data will be legal only if it matches, and the data will be cached, otherwise the data cannot be cached

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

Right link, source_business has several cached entries, source_order matches several cached entries.

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

Left link, source_order has several cached entries, source_business matches several cached entries.

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;

How to write time interval Join

Interval linking, data matching within a period of time

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