如何强制 Clickhouse 的 Kafka 引擎检查嵌套列的长度

Question

我有一个 Clickhouse 服务器，其中有一个带有嵌套字段和 kafka_handle_error_mode='stream',input_format_import_nested_json=1 设置的 Engine=Kafka 表和两个物化视图：

• 一个用于_error='' 的情况，它将数据存储到与Engine=Kafka 表结构相同的底层表中
• 一个用于_error!=''，用于存储原始消息和错误，以防出现“错误”数据

问题是，当 clickhouse 从 kafka 获取具有不同嵌套列长度（例如 {"n":{"a":["1","2"], "b":["3"]}}）的消息时，它会通过 Engine=Kafka 表而不生成 _error 并卡在表插入（并挂起整个保存循环），因为 Kafka 表不检查嵌套列的长度，但目标表会。

有一个flatten_nested=0 设置似乎改变了嵌套行为，但它需要不同的 json 结构，这对我来说是不可接受的。有解决办法吗？

Answer 1

我也很想知道是否有一种干净/高效的方法来解决它。

可能适合您的用例的建议是将字符串存储在 Kafka 表中。 在物化视图中，可以使用 Json Functions 和过滤错误...

CREATE TABLE kafka(
            payload String
            )  engine = Kafka
SETTINGS  kafka_broker_list = 'localhost:9092',
    kafka_topic_list = 'nested',
    kafka_group_name = 'nested',
    kafka_format = 'JSONAsString',
    kafka_num_consumers = 1,
    kafka_handle_error_mode = 'stream';

create materialized view  consumer
        to valid_payload_table
        as select
               JSONExtract(payload, 'a', 'String') as a, 
               JSONExtract(payload, 'b', 'String') as b,
        from kafka
        where _error!=''

Answer 2

1. Kafka 引擎不会检查嵌套数组的大小，因为这是 MergeTree 表的限制。

2. Kafka引擎的结构不一定要和MergeTree表匹配。只需在物化视图 SELECT 中添加相应的转换/检查即可。

例子：

create table n ( a Nested( n1 int, n2 int ) ) Engine=Kafka ....;
create table m ( an1 Array(int), an2 Array(int) ) Engine = MergeTree order by tuple();

create materialized view m_mv to m
as select 
     a.n1 as an1, 
     a.n2 as an2
from n;