Flink SQL反序列化失败时的DLQ处理方案咨询

一、跳过单条坏消息（纯SQL实现） #

在Kafka源表DDL中通过格式层配置开启容错，同时用SQL逻辑识别并分离坏消息：

• 开启解析容错：在FORMAT配置中添加'ignore-parse-errors' = 'true'，反序列化失败时字段会被设为NULL而非抛出异常，避免作业停服。
  示例DDL：

  CREATE TABLE kafka_source (
    id STRING,
    payload STRING
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'your_business_topic',
    'properties.bootstrap.servers' = 'xxx:9092',
    'format' = 'json',
    'ignore-parse-errors' = 'true',
    'scan.startup.mode' = 'group-offsets'
  );
  

• 分流处理：在查询中通过关键字段是否为NULL识别坏消息，正常消息走业务链路，坏消息写入DLQ存储（如专属Kafka主题）：

  -- 处理正常消息
  INSERT INTO business_sink
  SELECT id, payload FROM kafka_source WHERE id IS NOT NULL;
  
  -- 导出坏消息到DLQ
  INSERT INTO dlq_sink
  SELECT RAW('value') AS raw_message, CURRENT_TIMESTAMP() AS error_time 
  FROM kafka_source WHERE id IS NULL;
  

  其中RAW('value')可获取原始未解析的消息内容，便于后续排查问题。

二、应急跳过指定偏移量 #

若作业已因坏消息挂掉，可通过修改源表启动配置跳过故障偏移量：

1. 从Flink UI或作业日志定位失败的Kafka分区与偏移量。
2. 修改源表的启动模式为specific-offsets，指定跳过坏消息后的起始偏移量：

   CREATE TABLE kafka_source (
     -- 字段定义
   ) WITH (
     -- 其他配置
     'scan.startup.mode' = 'specific-offsets',
     'scan.startup.specific-offsets' = '{"your_topic":{"0":1000,"1":2000}}'
   );
   

   注意：该方式会跳过指定偏移量之前的消息，需确认跳过的消息无需处理或已完成兜底，避免违反exactly-once语义。

三、混合Table API实现Side Output #

若纯SQL灵活性不足，可结合Table API实现坏消息的Side Output，无需自定义连接器：

• 用Table API定义带Side Output的Kafka源：

  TableSource<Row> kafkaTableSource = KafkaTableSource.builder()
      .setBootstrapServers("xxx:9092")
      .setTopic("your_business_topic")
      .setDeserializer(new JsonRowDeserializationSchema.Builder(rowTypeInfo)
          .failOnMissingField(false)
          .withSideOutputTags(new OutputTag<String>("bad-records"){})
          .build())
      .build();
  
  tableEnv.registerTableSource("kafka_source", kafkaTableSource);
  

• SQL处理正常消息，Table API导出坏消息：

  // 业务逻辑SQL
  tableEnv.sqlUpdate("INSERT INTO business_sink SELECT * FROM kafka_source");
  
  // 将Side Output的坏消息写入DLQ
  DataStream<String> badRecords = ((StreamTableSource<?>) kafkaTableSource)
      .getDataStream(env)
      .getSideOutput(new OutputTag<String>("bad-records"){});
  badRecords.addSink(new KafkaSink<>("dlq_topic", new SimpleStringSchema(), props));
  

四、优化重消费耗时（缓解SLA压力） #

若无法避免作业重启，可通过以下配置缩短恢复时间：

• 开启RocksDB增量检查点：在flink-conf.yaml中设置state.backend.rocksdb.incremental-checkpoint.enabled: true，减少检查点写入量。
• 启用本地恢复：设置state.backend.local-recovery: true，利用本地存储恢复状态，避免全量拉取分布式存储的状态数据。
• 调优检查点间隔：根据业务容忍度适当增大检查点间隔，减少检查点开销，同时控制故障回滚范围。

内容的提问来源于stack exchange，提问作者Robin F.