麻烦问一下Flink CDC假如create_time是敏感字段我需要把id和name同步过来？

在使用Flink CDC进行数据同步时，如果需要过滤掉某些字段（如create_time），但发现过滤未生效，可能是由于以下几个原因导致的。以下是详细分析和解决方案：

1. 字段过滤的实现方式

Flink CDC本身并不直接支持在源表中对字段进行过滤。字段过滤通常需要通过以下两种方式实现： - SQL投影：在Flink SQL中通过SELECT语句选择需要的字段。 - 自定义反序列化器：在DataStream API中实现自定义的反序列化逻辑。

如果您仅在配置中尝试过滤字段（例如通过tableList或scan.startup.mode等参数），这种方式是无效的，因为这些参数主要用于控制表级别的读取行为，而不是字段级别的过滤。

2. 正确的字段过滤方法

方法一：使用SQL投影

在Flink SQL中，您可以通过SELECT语句明确指定需要同步的字段。例如：

CREATE TABLE source_table (
    id INT,
    name STRING,
    create_time TIMESTAMP
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<your-host>',
    'port' = '<your-port>',
    'username' = '<your-username>',
    'password' = '<your-password>',
    'database-name' = '<your-database>',
    'table-name' = '<your-table>'
);

CREATE TABLE sink_table (
    id INT,
    name STRING
) WITH (
    'connector' = 'hologres',
    'hostname' = '<your-hologres-host>',
    'port' = '<your-hologres-port>',
    'username' = '<your-hologres-username>',
    'password' = '<your-hologres-password>',
    'database-name' = '<your-hologres-database>',
    'table-name' = '<your-hologres-table>'
);

INSERT INTO sink_table
SELECT id, name
FROM source_table;

上述SQL语句中，SELECT id, name明确指定了只同步id和name字段，而忽略了create_time字段。

方法二：使用自定义反序列化器

如果您使用的是DataStream API，可以通过实现DebeziumDeserializationSchema来自定义反序列化逻辑。例如：

public class CustomDeserializer implements DebeziumDeserializationSchema<RowData> {
    @Override
    public void deserialize(SourceRecord record, Collector<RowData> out) throws Exception {
        // 解析SourceRecord并提取需要的字段
        Struct value = (Struct) record.value();
        int id = value.getInt32("id");
        String name = value.getString("name");

        // 构造RowData对象，忽略create_time字段
        RowData rowData = new RowData(id, name);
        out.collect(rowData);
    }
}

然后在Flink作业中使用该自定义反序列化器：

MySQLSource<RowData> mySQLSource = MySQLSource.<RowData>builder()
    .hostname("<your-host>")
    .port(<your-port>)
    .databaseList("<your-database>")
    .tableList("<your-table>")
    .username("<your-username>")
    .password("<your-password>")
    .deserializer(new CustomDeserializer())
    .build();

3. 可能的原因与排查步骤

原因一：字段过滤未正确配置

如果您尝试通过其他方式（如WITH参数）过滤字段，这通常是无效的，因为Flink CDC的WITH参数主要用于控制连接器的行为，而不是字段级别的过滤。

原因二：下游表结构不匹配

即使在SQL中进行了字段过滤，如果下游表的结构仍然包含create_time字段，可能会导致写入失败。请确保下游表的结构与上游过滤后的字段一致。

原因三：全量阶段未过滤字段

Flink CDC在全量读取阶段会读取表中的所有字段，即使在增量阶段进行了字段过滤，全量阶段的数据仍可能包含create_time字段。为避免此问题，可以在全量阶段也通过SQL投影或自定义反序列化器进行字段过滤。

4. 重要提醒

• 字段过滤需显式声明：无论是通过SQL投影还是自定义反序列化器，字段过滤都需要显式声明，不能依赖默认行为。
• 检查上下游表结构：确保下游表的结构与过滤后的字段一致，否则可能导致写入失败。
• 全量与增量阶段一致性：全量和增量阶段的字段过滤逻辑应保持一致，以避免数据不一致问题。

5. 总结与建议

要解决create_time字段未被过滤的问题，建议按照以下步骤操作： 1. 使用SQL投影明确指定需要同步的字段。 2. 如果使用DataStream API，实现自定义反序列化器以过滤字段。 3. 检查上下游表结构是否一致。 4. 确保全量和增量阶段的字段过滤逻辑一致。

通过以上方法，您可以有效过滤掉不需要的字段，并确保数据同步的正确性。