SQL Server real-time synchronization

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

Real-time synchronization of data from the result table to the target table in SQL Server

Case

Synchronize data in cdc_source to cdc_target in SQL Server in real time

Premise

• Start table cdc of SQL Server

Specify the cdc_source table to enable 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;

solution

Create a Flink SQL job type and add the following dependencies

• [flink-sql-connector-SQL Server-cdc-3.2.1.jar download](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 download](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;