Flink CDC里SQLserver CDC，有什么排查思路吗？

在Flink CDC中，SQL Server CDC（Change Data Capture）可以通过捕获数据库的变更日志来获取数据的插入（INSERT）、更新（UPDATE）和删除（DELETE）操作。以下是关于如何使用Flink SQL捕获SQL Server中的INSERT操作的详细说明：

1. SQL Server CDC的基本原理

SQL Server CDC通过读取事务日志（Transaction Log）来捕获数据的变化。启用CDC后，SQL Server会将表的变更记录存储在特定的系统表中，这些记录包括INSERT、UPDATE和DELETE操作的具体信息。

Flink通过集成CDC工具（如Debezium或自定义的SQL Server CDC连接器），可以实时读取这些变更日志，并将其解析为Flink SQL中的Changelog事件流。其中，INSERT操作会被解析为RowKind.INSERT类型的事件。

2. 配置SQL Server CDC源表

为了捕获SQL Server中的INSERT操作，您需要在Flink SQL中创建一个CDC源表。以下是一个示例DDL语句：

CREATE TABLE sqlserver_source (
    id BIGINT,
    name STRING,
    description STRING,
    weight DECIMAL(10, 2),
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'sqlserver-cdc',
    'hostname' = 'your-sqlserver-host',
    'port' = '1433',
    'username' = 'your-username',
    'password' = 'your-password',
    'database-name' = 'your-database',
    'schema-name' = 'dbo',
    'table-name' = 'your-table',
    'scan.startup.mode' = 'initial'
);

关键参数说明： - connector: 指定为sqlserver-cdc，表示使用SQL Server CDC连接器。 - hostname 和 port: SQL Server实例的地址和端口。 - database-name 和 table-name: 指定要捕获变更的目标数据库和表。 - scan.startup.mode: 定义初始同步模式，可选值包括initial（全量+增量）和latest-offset（仅增量）。

3. 捕获INSERT操作

在Flink SQL中，SQL Server CDC源表会生成包含RowKind的Changelog事件流。对于INSERT操作，Flink会生成RowKind.INSERT类型的事件。例如：

SELECT * FROM sqlserver_source WHERE op_type = '+I';

说明： - op_type 是元数据列，用于标识每条记录的操作类型： - +I 表示INSERT操作。 - -D 表示DELETE操作。 - -U 和 +U 分别表示UPDATE_BEFORE和UPDATE_AFTER操作。

4. 处理INSERT事件

捕获到INSERT事件后，您可以将其与其他数据流进行关联、聚合或写入下游存储。例如，将INSERT事件写入Kafka：

CREATE TABLE kafka_sink (
    id BIGINT,
    name STRING,
    description STRING,
    weight DECIMAL(10, 2)
) WITH (
    'connector' = 'kafka',
    'topic' = 'insert-events',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

INSERT INTO kafka_sink
SELECT * FROM sqlserver_source WHERE op_type = '+I';

5. 注意事项

• CDC功能启用: 在SQL Server中，必须先为目标表启用CDC功能。可以通过以下SQL命令启用：

  EXEC sys.sp_cdc_enable_table
    @source_schema = 'dbo',
    @source_name = 'your-table',
    @role_name = NULL;
  

  启用后，SQL Server会在系统表中记录该表的所有变更。

• Flink版本要求: 确保使用的Flink版本支持SQL Server CDC连接器。建议使用VVR 6.0及以上版本。

• 性能优化: 如果数据量较大，建议调整Flink任务的并行度和SQL Server的CDC捕获频率，以避免性能瓶颈。

通过上述配置和操作，您可以成功捕获SQL Server中的INSERT操作，并将其应用于实时数据处理场景中。