Flink SQL中Kafka Source如何跳过失败消息？

方案1：利用JSON格式内置错误处理参数 #

Flink的JSON格式支持直接配置解析错误的处理策略，通过添加以下参数即可跳过无效消息：

• 'json.ignore-parse-errors' = 'true'：遇到解析失败的消息时跳过，解析失败的字段会被设为NULL
• 'json.fail-on-missing-field' = 'false'：忽略字段缺失的情况，避免因字段不匹配触发异常

修改后的Kafka源表完整配置示例：

CREATE TABLE kafka_source (
  id INT,
  name STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'test',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true',
  'json.fail-on-missing-field' = 'false',
  'properties.bootstrap.servers' = 'localhost:9092',
  'scan.startup.mode' = 'latest-offset'
);

方案2：自定义格式实现DLQ转发 #

如果需要将无效消息转发到死信队列（DLQ），可以自定义Flink Format：

1. 实现DeserializationSchema接口，在deserialize方法中捕获解析异常
2. 将无效消息发送到指定的Kafka DLQ主题
3. 在Flink SQL中注册该自定义格式并引用

核心Java实现示例：

public class CustomJsonDLQFormat implements DeserializationSchema<RowData> {
    private final JsonRowDataDeserializationSchema delegate;
    private final KafkaProducer<String, String> dlqProducer;
    private final String dlqTopic;

    public CustomJsonDLQFormat(JsonRowDataDeserializationSchema delegate, String dlqTopic) {
        this.delegate = delegate;
        this.dlqTopic = dlqTopic;
        this.dlqProducer = new KafkaProducer<>(getDLQProducerConfig());
    }

    @Override
    public RowData deserialize(byte[] message) throws IOException {
        try {
            return delegate.deserialize(message);
        } catch (Exception e) {
            // 发送无效消息到DLQ
            dlqProducer.send(new ProducerRecord<>(dlqTopic, new String(message)));
            return null; // 返回null表示跳过当前消息
        }
    }

    // 实现其他接口方法及配置初始化逻辑...
}

注册自定义格式后，即可在CREATE TABLE的format参数中引用。

方案3：混合DataStream API与SQL处理侧输出流 #

纯SQL无法满足复杂需求时，可以结合DataStream API实现侧输出流分离无效消息：

1. 用DataStream API读取Kafka流，在处理逻辑中捕获解析异常，将无效消息输出到侧输出流
2. 将正常消息流转换为临时视图，供Flink SQL查询
3. 单独处理侧输出流中的无效消息（写入DLQ或存储）

示例代码片段：

// 定义侧输出流标签
OutputTag<String> invalidMsgTag = new OutputTag<String>("invalid-kafka-messages"){};

// 读取Kafka流并分离有效/无效消息
DataStream<String> validStream = env.addSource(new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), kafkaProps))
    .process(new ProcessFunction<String, String>() {
        @Override
        public void processElement(String value, Context ctx, Collector<String> out) {
            try {
                // 验证JSON格式
                JSONObject.parseObject(value);
                out.collect(value);
            } catch (JSONException e) {
                // 发送到侧输出流
                ctx.output(invalidMsgTag, value);
            }
        }
    });

// 将有效流注册为SQL视图
Table validTable = tableEnv.fromDataStream(validStream, Schema.newBuilder()
    .column("raw_data", DataTypes.STRING())
    .build());
tableEnv.createTemporaryView("valid_kafka_source", validTable);

// 用SQL查询有效数据
tableEnv.executeSql("SELECT JSON_VALUE(raw_data, '$.id') AS id FROM valid_kafka_source");

// 处理无效消息，写入DLQ
validStream.getSideOutput(invalidMsgTag)
    .addSink(new FlinkKafkaProducer<>("kafka-dlq-topic", new SimpleStringSchema(), dlqProps));

内容的提问来源于stack exchange，提问作者oceansize