Flink流平台-实践sqlServer单数据源CDC之双流join

背景准备

曾经在Flink CDC同步mysql和pg关联数据到es实践尝试使用flink进行多数据源双流join到es，本次实践使用流平台单数据源双流join。

我们快速用docker搭建起来sqlserver,

记住开启sqlserver cdc:sys.sp_cdc_enable_table

docker-compose

version: '2.1'
services:
   sqlserver:
     image: mcr.microsoft.com/mssql/server:2019-latest
     container_name: sqlserver
     ports:
       - "1433:1433"
     environment:
       - "MSSQL_AGENT_ENABLED=true"
       - "MSSQL_PID=Standard"
       - "ACCEPT_EULA=Y"
       - "SA_PASSWORD=Password!"

并创建一些数据

-- Sqlserver
CREATE DATABASE inventory;
GO
USE inventory;
EXEC sys.sp_cdc_enable_db;

-- Create and populate our products using a single insert with many rows
CREATE TABLE products (
id INTEGER IDENTITY(101,1) NOT NULL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description VARCHAR(512),
weight FLOAT
);
INSERT INTO products(name,description,weight)
VALUES ('scooter','Small 2-wheel scooter',3.14);
INSERT INTO products(name,description,weight)
VALUES ('car battery','12V car battery',8.1);
INSERT INTO products(name,description,weight)
VALUES ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8);
INSERT INTO products(name,description,weight)
VALUES ('hammer','12oz carpenter''s hammer',0.75);
INSERT INTO products(name,description,weight)
VALUES ('hammer','14oz carpenter''s hammer',0.875);
INSERT INTO products(name,description,weight)
VALUES ('hammer','16oz carpenter''s hammer',1.0);
INSERT INTO products(name,description,weight)
VALUES ('rocks','box of assorted rocks',5.3);
INSERT INTO products(name,description,weight)
VALUES ('jacket','water resistent black wind breaker',0.1);
INSERT INTO products(name,description,weight)
VALUES ('spare tire','24 inch spare tire',22.2);
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'products', @role_name = NULL, @supports_net_changes = 0;
-- Create some very simple orders
CREATE TABLE orders (
id INTEGER IDENTITY(10001,1) NOT NULL PRIMARY KEY,
order_date DATE NOT NULL,
purchaser INTEGER NOT NULL,
quantity INTEGER NOT NULL,
product_id INTEGER NOT NULL,
FOREIGN KEY (product_id) REFERENCES products(id)
);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('16-JAN-2016', 1001, 1, 102);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('17-JAN-2016', 1002, 2, 105);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('19-FEB-2016', 1002, 2, 106);
INSERT INTO orders(order_date,purchaser,quantity,product_id)
VALUES ('21-FEB-2016', 1003, 1, 107);
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'orders', @role_name = NULL, @supports_net_changes = 0;
GO

使用流平台快速搭建数据流

本次我们从sqlserver获取数据，并通过print直接输出到终端

CREATE TABLE products (
  id INT,
  name STRING,
  description STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname' = '192.168.31.6',
  'port' = '1433',
  'username' = 'sa',
  'password' = 'Password!',
  'database-name' = 'inventory',
  'schema-name' = 'dbo',
  'table-name' = 'products'
);


CREATE TABLE orders (
  id INT,
  order_date DATE,
  purchaser INT,
  quantity INT,
  product_id INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname' = '192.168.31.6',
  'port' = '1433',
  'username' = 'sa',
  'password' = 'Password!',
  'database-name' = 'inventory',
  'schema-name' = 'dbo',
  'table-name' = 'orders'
);


CREATE TABLE enriched_orders (
  order_id INT,
  order_date DATE,
  purchaser INT,
  quantity INT,
  product_name STRING,
  product_description STRING
) WITH ('connector' = 'print');


INSERT INTO
  enriched_orders
SELECT
  o.id,
  o.order_date,
  o.purchaser,
  o.quantity,
  p.name,
  p.description
FROM
  orders AS o
  LEFT JOIN products AS p ON o.product_id = p.id;

启动sqlServer_cdc_print