将Kafka中JSON数组消息解析到ClickHouse 22.8.1.1时遇解析异常

核心问题成因 #

你遇到的解析错误，是因为kafka_format='JSONAsString'要求每条Kafka消息必须是完整有效的JSON值（比如单个JSON对象或闭合的JSON数组），但实际消费的消息中存在不完整的JSON结构——从错误信息能看到，消息内容出现了多个JSON对象拼接（...}, {"partner_id":156...），却没有正确的数组闭合符]，导致ClickHouse解析JSON时失败。

解决方案 #

方案1：直接存储原始消息（跳过JSON解析） #

如果只是想先把Kafka消息完整存入ClickHouse，后续再处理解析，改用Raw格式即可——这个格式会直接将消息原始内容作为字符串存储，不会尝试解析JSON：

CREATE TABLE testdb.queue_json (
        `message` String
) 
ENGINE = Kafka 
SETTINGS 
        kafka_broker_list = 'xxx', 
        kafka_topic_list = 'mytopic', 
        kafka_group_name = 'mygroup1', 
        kafka_format = 'Raw',
        kafka_max_block_size = 1048576

方案2：解析JSON数组为结构化数据 #

如果需要直接将JSON数组中的对象解析为结构化数据，可以创建物化视图配合处理：

1. 先按方案1用Raw格式创建Kafka引擎表；
2. 创建物化视图拆分JSON数组并提取字段：

CREATE MATERIALIZED VIEW testdb.parsed_queue_json 
ENGINE = MergeTree()
ORDER BY id
AS
SELECT 
    JSONExtract(obj, 'partner_id', 'UInt32') AS partner_id,
    JSONExtract(obj, 'id', 'String') AS id,
    JSONExtract(obj, 'type', 'UInt8') AS type,
    JSONExtract(obj, 'remove', 'UInt8') AS remove,
    JSONExtract(obj, 'segments', 'Array(String)') AS segments
FROM testdb.queue_json,
JSONExtractArrayElements(message) AS obj

额外注意事项 #

• 检查Kafka生产者逻辑：确保每条消息是完整闭合的JSON数组，避免出现消息拆分、多个对象拼接无数组包裹的情况；
• 如果生产者可以调整为单条消息发送单个JSON对象，也可以直接用kafka_format='JSONEachRow'创建Kafka表，直接解析为结构化数据，无需额外处理数组。

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