Flink CDC 可以读取虚拟列op吗？

Flink CDC 可以读取虚拟列 op，但是需要使用自定义的 DebeziumDeserializationSchema 来解析该列。

具体来说，您需要在 DebeziumDeserializationSchemaBuilder 中添加一个名为 op 的列，并指定其数据类型和反序列化器。例如：

DebeziumDeserializationSchema<MyEvent> deserializationSchema = DebeziumDeserializationSchemaBuilder<MyEvent>.create()
    .schemaName("my_schema")
    .databaseList(Arrays.asList("mydb"))
    .tableList(Arrays.asList("mytable"))
    .column("id", DataTypes.BIGINT())
    .column("name", DataTypes.STRING())
    .column("op", DataTypes.STRING()) // 添加虚拟列 op
    .deserializer(new MyEventDeserializer())
    .build();

然后，您可以在您的应用程序中使用 FlinkCDC 连接器来读取数据流，并根据 op 列的值执行不同的操作。例如：

FlinkCDC<MyEvent> cdc = new FlinkCDC<>(env, mySourceFunction, deserializationSchema);
DataStream<MyEvent> stream = env.addSource(cdc);
stream.filter(event -> event.getOp().equals("INSERT")) // 根据 op 列的值过滤数据流
      .map(event -> { /* 对 INSERT 事件执行某些操作 */ return event; })
      .otherwise(event -> { /* 对其他事件执行某些操作 */ return event; });

Flink CDC 从 MySQL 中读取的数据变更事件中包含了一个名为 _op 的虚拟列，它表示了数据的变更类型。这个列包含了三种可能的值：I（插入）、U（更新）和 D（删除）。这些值可以用来决定在处理数据时进行不同的操作。

要使用 Flink SQL 来根据 _op 列的不同值执行不同的操作，你可以使用 CASE WHEN 或 UNION ALL 结构来实现。例如，以下是一个简化的示例：

-- 创建一个临时表以接收CDC事件
CREATE TABLE mysql_events (
    id INT,
    name STRING,
    price DECIMAL(10,2),
    _op STRING
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'password',
    'database-name' = 'test',
    'table-name' = 'products'
);

-- 根据_op列的不同值进行不同的操作
INSERT INTO sink_table
SELECT 
    id,
    name,
    CASE 
        WHEN _op = 'I' THEN ...
        WHEN _op = 'U' THEN ...
        WHEN _op = 'D' THEN ...
    END as new_price
FROM mysql_events;

在这个例子中，你可以在每个 WHEN 子句中定义对应的操作。注意，你需要为每个操作提供完整的语句，包括字段名、运算符等。