Flink CDC (Change Data Capture) 是 Apache Flink 的一个组件，它允许你捕获数据库表中的变更事件。对于 SQL Server 数据库，Flink CDC 支持通过 Debezium 连接器来捕获变更事件。
在 Flink CDC 中，op 字段通常代表操作类型，比如 INSERT, UPDATE, DELETE 等。当你使用 Flink CDC 从 SQL Server 捕获变更数据时，op 字段会被自动包含在捕获的事件中。
如何配置 Flink CDC 从 SQL Server 捕获变更数据
1.添加依赖: 在你的项目中添加 Flink CDC 的依赖。对于 SQL Server，你需要添加Debezium连接器的依赖。
如果你使用的是 Maven，可以在 pom.xml 文件中添加如下依赖：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-debezium_2.12</artifactId>
<version>1.16.0</version> <!-- 根据你的Flink版本选择合适的版本 -->
</dependency>
2.配置 Flink CDC: 你需要配置 Flink CDC 的 Source 连接器来从 SQL Server 捕获变更数据。这可以通过 Flink SQL 或者通过编写 Java/Scala 代码来完成。
使用 Flink SQL 配置
CREATE TABLE sql_server_source (
id INT,
name STRING,
-- 其他列...
op STRING, -- 这个字段会自动包含操作类型
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'debezium',
'debezium.catalog-name' = 'sqlserver-catalog', -- 必须与配置文件中的catalog.name一致
'debezium.database.hostname' = 'localhost',
'debezium.database.port' = '1433',
'debezium.database.user' = 'your_user',
'debezium.database.password' = 'your_password',
'debezium.database.dbname' = 'your_database',
'debezium.table.whitelist' = 'your_schema.your_table',
'debezium.snapshot.locking.mode' = 'none', -- 避免锁表
'debezium.include.schema.changes' = 'true'
);
使用 Java/Scala 配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkDebeziumSource<Row> source = FlinkDebeziumSource.forInstance(
new MySqlSourceBuilder()
.hostname("localhost")
.port(1433)
.databaseList("your_database")
.tableList("your_schema.your_table")
.username("your_user")
.password("your_password")
.deserializer(new JdbcRowDeserializationSchema.Builder()
.typeInfo(TypeInformation.of(Row.class))
.build())
.build(),
env
);
DataStream<Row> stream = env.addSource(source);
在这个例子中，op 字段会自动包含在捕获的事件中，你可以在后续的 SQL 查询或者数据流处理中直接使用它。
示例查询
一旦你配置好了 Flink CDC，并且开始捕获 SQL Server 的变更数据，你可以使用如下 SQL 查询来获取 op 字段：
SELECT op, id, name, -- 以及其他你需要的字段
FROM sql_server_source;
这里 op 字段代表了变更事件的操作类型。你可以根据需要进一步处理这些数据，例如过滤特定的操作类型或聚合数据。
