有没有人遇到Flink CDC1.16 解析json乱码的问题？

在使用 Flink CDC 1.16 解析 JSON 数据时，如果遇到乱码问题，可能是由于以下几个原因导致的。以下是可能的原因分析及解决方案：

1. 字符编码不一致

Flink 在解析 JSON 数据时，默认会使用 UTF-8 编码。如果数据源（如 Kafka、MySQL 等）中的数据编码格式不是 UTF-8，则可能导致解析后的数据出现乱码。

解决方案：

• 确保数据源的编码格式为 UTF-8。
• 如果无法修改数据源的编码格式，可以在 Flink 中显式指定编码方式。例如，在使用 JsonDebeziumDeserializationSchema 时，可以通过配置 Properties 来指定编码格式：

  Properties properties = new Properties();
  properties.setProperty("database.encoding", "UTF-8"); // 或其他编码格式
  MySqlSource.<String>builder()
    .hostname(config.getHostname())
    .port(config.getPort())
    .databaseList(config.getDatabaseList())
    .tableList(config.getTableList())
    .username(config.getUsername())
    .password(config.getPassword())
    .deserializer(new JsonDebeziumDeserializationSchema())
    .debeziumProperties(properties)
    .build();
  

2. JSON 格式解析错误

如果 JSON 数据本身存在格式问题（如字段缺失、嵌套结构未正确解析等），可能会导致解析失败或乱码。

解决方案：

• 检查 JSON 数据是否符合标准格式。可以使用工具（如在线 JSON 校验器）验证数据的合法性。
• 在 Flink 的 JSON 配置中启用容错选项，避免因解析错误导致作业失败：

  CREATE TABLE kafka_source (
    id INT,
    name STRING,
    info MAP<STRING, STRING>
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false', -- 忽略缺失字段
    'json.ignore-parse-errors' = 'true'    -- 忽略解析错误
  );
  

3. 数值类型被解析为字符串

在使用 JsonDebeziumDeserializationSchema 时，如果数值类型（如 BIGINT、DECIMAL）被解析为字符串，可能会导致数据展示异常或乱码。

解决方案：

• 配置 Debezium 的相关参数，确保数值类型被正确解析：

  Properties properties = new Properties();
  properties.setProperty("bigint.unsigned.handling.mode", "long");
  properties.setProperty("decimal.handling.mode", "double");
  
  MySqlSource.<String>builder()
    .hostname(config.getHostname())
    .port(config.getPort())
    .databaseList(config.getDatabaseList())
    .tableList(config.getTableList())
    .username(config.getUsername())
    .password(config.getPassword())
    .deserializer(new JsonDebeziumDeserializationSchema())
    .debeziumProperties(properties)
    .build();
  

4. Kafka 消息键值解析冲突

如果 Kafka 消息的 Key 和 Value 中存在同名字段，可能会导致解析冲突，进而引发乱码问题。

解决方案：

• 使用 key.fields-prefix 和 value.fields-prefix 参数为 Key 和 Value 字段添加前缀，避免命名冲突：

  CREATE TABLE kafka_source (
    key_id INT,
    value_name STRING
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'key.fields-prefix' = 'key_',       -- Key 字段前缀
    'value.fields-prefix' = 'value_'    -- Value 字段前缀
  );
  

5. 时间戳格式不匹配

如果 JSON 数据中的时间戳格式与 Flink 默认的时间戳格式（yyyy-MM-dd HH:mm:ss.s{precision}）不一致，可能会导致解析失败或乱码。

解决方案：

• 配置 json.timestamp-format.standard 参数，指定时间戳格式：

  CREATE TABLE kafka_source (
    event_time TIMESTAMP(3)
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.timestamp-format.standard' = 'ISO-8601' -- 或 'SQL'
  );
  

6. 依赖版本问题

如果使用的 Flink CDC 版本与连接器版本不匹配，可能会导致解析异常。

解决方案：

• 确保使用的 Flink CDC 连接器版本与 Flink 引擎版本兼容。例如，Flink CDC 1.16 应搭配 Flink 1.16 使用。
• 如果使用的是 SNAPSHOT 版本，请确认 Maven 仓库路径是否正确。从 2.x 版本开始，Flink CDC 的 Group ID 已从 com.alibaba.ververica 改为 com.ververica。

7. 其他注意事项

• 检查日志：查看 Flink 作业日志，定位具体的错误信息，有助于快速排查问题。
• 升级版本：如果问题持续存在，建议升级到最新版本的 Flink CDC，以获取最新的修复和功能支持。

通过以上步骤，您可以有效解决 Flink CDC 1.16 解析 JSON 数据时的乱码问题。如果问题仍未解决，请提供更多上下文信息（如 JSON 数据样例、配置文件内容等），以便进一步分析。