如何使用Flink SQL读取Kafka中的空值消息（Flink 1.18）

默认情况下，Flink Kafka连接器会跳过value为null的消息，同时JSON格式解析器无法直接处理空消息体，导致这类消息被丢弃。要读取这类消息，可通过以下两种方案调整配置：

方案一：直接读取为全null行 #

修改表定义，添加skip-null-values参数并设置为false，关闭连接器跳过null消息的行为。此时消息体为null时，表中所有字段会被解析为null：

CREATE TABLE kafka_table (
  field1 INT,
  field2 STRING,
  field3 DOUBLE
) WITH (
  'connector' = 'kafka',
  'topic' = 'your_topic_name',
  'properties.bootstrap.servers' = 'your_bootstrap_servers',
  'properties.group.id' = 'your_consumer_group_id',
  'format' = 'json',
  'skip-null-values' = 'false'
);

方案二：区分消息体为null和JSON全null的情况 #

如果需要明确判断消息体本身是否为null（而非JSON内容全为null），可以先用RAW格式接收原始字节，再通过视图解析：

1. 创建RAW格式的Kafka表，接收原始消息字节：

CREATE TABLE kafka_raw_table (
  raw_value BYTES
) WITH (
  'connector' = 'kafka',
  'topic' = 'your_topic_name',
  'properties.bootstrap.servers' = 'your_bootstrap_servers',
  'properties.group.id' = 'your_consumer_group_id',
  'format' = 'raw',
  'skip-null-values' = 'false'
);

2. 创建视图解析JSON并标记null消息：

CREATE VIEW kafka_parsed_view AS
SELECT
  CASE
    WHEN raw_value IS NULL THEN NULL
    ELSE JSON_VALUE(raw_value, '$.field1')::INT
  END AS field1,
  CASE
    WHEN raw_value IS NULL THEN NULL
    ELSE JSON_VALUE(raw_value, '$.field2')::STRING
  END AS field2,
  CASE
    WHEN raw_value IS NULL THEN NULL
    ELSE JSON_VALUE(raw_value, '$.field3')::DOUBLE
  END AS field3,
  raw_value IS NULL AS is_null_message
FROM kafka_raw_table;

通过is_null_message字段可直接判断当前消息是否为体为null的情况。

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