SQL Server实时同步

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

实现在SQL Server中结果表向目标表中实时同步数据

案例

SQL Server中将cdc_source中的数据实时同步到cdc_target中

前提

• 启动SQL Server的表cdc

指定cdc_source表开启cdc

create database ispong_db;

-- 启用数据库的CDC功能
USE ispong_db;
EXEC sys.sp_cdc_enable_db;

-- 查看数据库CDC是否已经启用，ispong_db库
SELECT name, is_cdc_enabled
FROM sys.databases
WHERE name = 'ispong_db';

-- 启用表的CDC功能
USE ispong_db;
EXEC sys.sp_cdc_enable_table
     @source_schema = N'dbo',           
     @source_name = N'cdc_source',        
     @role_name = NULL;                   

-- 查看表CDC是否已经启用
use ispong_db;
SELECT * FROM cdc.change_tables;

解决方案

创建Flink SQL作业类型，添加以下依赖

• [flink-sql-connector-SQL Server-cdc-3.2.1.jar下载](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-SQL Server-cdc/3.2.1/flink-sql-connector-SQL Server-cdc-3.2.1.jar)
• [mssql-JDBC-11.2.3.jre8.jar下载](https://repo1.maven.org/maven2/com/microsoft/SQL Server/mssql-JDBC/11.2.3.jre8/mssql-JDBC-11.2.3.jre8.jar)

CREATE TABLE from_table
(
    username STRING,
    age      INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
      'connector' = 'sqlserver-cdc',
      'hostname' = '147.192.168.116',
      'port' = '30103',
      'username' = 'sa',
      'password' = 'ispong123',
      'database-name' = 'ispong_db',
      'table-name' = 'dbo.cdc_source2',
      'debezium.database.trustServerCertificate' = 'true'
      );

CREATE TABLE target_table
(
    username STRING,
    age      INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:sqlserver://147.192.168.116:30103;database=ispong_db;trustServerCertificate=true',
      'driver' = 'com.microsoft.sqlserver.jdbc.SQLServerDriver',
      'table-name' = 'cdc_sink',
      'username' = 'SA',
      'password' = 'ispong123',
      'sink.buffer-flush.max-rows' = '1'
      );

INSERT INTO target_table
select *
from from_table;