ByConity Kafka引擎数据写入问题排查与解决方案

问题背景

在使用ByConity 0.4.2版本部署的K8s环境中，用户遇到了Kafka引擎表无法正常写入数据的问题。虽然Kafka消费者独立Pod中能看到数据消费的日志，但存储表中始终没有数据写入。通过kafka-console-consumer工具验证，确认Kafka主题中的数据格式是正确的JSON格式。

问题排查过程

1. 日志分析：

   • 检查cnch_system.cnch_kafka_log表，发现只有POLL和COMMIT类型的事件记录，没有PARSE_ERROR类型的错误日志
   • Kafka消费者Pod中只有Debug级别的日志："IRowInputFormat: Skipped 4 rows with errors while reading the input stream"

2. 数据验证：

   • 确认Kafka主题中确实有数据生产，且其他Flink任务能正常消费
   • 通过kafka-console-consumer验证数据格式为有效的JSON

3. 配置检查：

   • 存储表使用了不同的虚拟仓库(vw)
   • 消费表配置了DEFAULT列

根本原因分析

经过深入排查，发现问题源于JSON数据的严格解析机制：

1. JSON解析严格性：

   • ByConity新版本采用了更严格的JSON解析逻辑
   • 当JSON中包含嵌套对象时，如果表字段定义为String类型，默认情况下会解析失败

2. 日志记录缺陷：

   • 当Kafka表配置了DEFAULT列时，PARSE_ERROR类型的错误日志不会被记录到cnch_kafka_log表中
   • 这是ByConity的一个已知问题，开发团队已计划修复

解决方案

1. 配置调整：

   -- 在消费的虚拟仓库(vw)上添加以下配置
   SETTINGS input_format_json_read_objects_as_strings = true
   

   这个配置允许将JSON对象作为字符串整体解析，而不是尝试解析其内部结构。

2. 临时规避方案：

   • 如果不使用DEFAULT列，错误日志能够正常记录
   • 但这只是临时方案，最佳实践还是调整JSON解析配置

高级应用技巧

1. 保存原始JSON内容： 如果需要同时保存原始JSON内容和解析后的字段，可以使用_content虚拟列：

   -- 物化视图表定义示例
   CREATE MATERIALIZED VIEW mv_table
   ENGINE = CnchMergeTree
   ORDER BY (dt, insert_time)
   AS SELECT 
       _content AS raw_json,
       topic,
       insert_time,
       dt
   FROM kafka_consume_table;
   

2. 性能调优参数：

   • kafka_max_block_size：控制每次从Kafka读取的最大数据量
   • kafka_poll_timeout_ms：设置轮询超时时间
   • kafka_flush_interval_ms：控制数据刷新的间隔时间

注意事项

1. 通配符限制：

   • kafka_topic_list参数不支持通配符(如pre.*)匹配多个topic
   • 需要明确列出所有要消费的topic名称

2. 容错机制：

   • Kafka消费位点信息会保存在Catalog中
   • 即使消费专用的虚拟仓库(vw)所在宿主机宕机，恢复后也能从正确位置继续消费

3. 监控建议：

   • 定期检查cnch_kafka_log表中的POLL事件metric值，确认实际消费的数据量
   • 监控COMMIT事件，确保消费进度正常更新

总结

ByConity的Kafka引擎在数据消费方面提供了强大的功能，但在实际使用中需要注意JSON解析的严格性和相关配置。通过合理设置input_format_json_read_objects_as_strings参数，可以解决大多数JSON格式解析问题。同时，了解系统的日志记录机制和容错特性，有助于构建更健壮的数据管道。