ClickHouse JSON解析异常：无法解析输入：预期'，'之前答案

【问题标题】：ClickHouse JSON parse exception: Cannot parse input: expected ',' beforeClickHouse JSON解析异常：无法解析输入：预期'，'之前
【发布时间】：2021-03-14 04:16:04
【问题描述】：

我正在尝试将 JSON 数据从 Kafka 添加到 ClickHouse。这是简化的 JSON：

{
  ...
   "sendAddress":{
      "sendCommChannelTypeId":4,
      "sendCommChannelTypeCode":"SMS",
      "sendAddress":"789345345945"},
   ...
}

这是在ClickHouse中创建表的步骤，使用Kafka Engine创建另一个表并创建MATERIALIZED VIEW来连接这两个表，并将CH与Kafka连接。

创建第一个表

CREATE TABLE tab 
(
    ...

    sendAddress Tuple (sendCommChannelTypeId Int32, sendCommChannelTypeCode String, sendAddress String),
     ...

)Engine = MergeTree()
PARTITION BY applicationId
ORDER BY (applicationId);

使用 Kafka Engine SETTINGS 创建第二个表：

CREATE TABLE tab_kfk
(
    ...
    sendAddress Tuple (sendCommChannelTypeId Int32, sendCommChannelTypeCode String, sendAddress String),
    ...
)ENGINE = Kafka
SETTINGS kafka_broker_list = 'localhost:9092',
       kafka_topic_list = 'topk2',
       kafka_group_name = 'group1',
       kafka_format = 'JSONEachRow',
       kafka_row_delimiter = '\n';

创建实体化视图

CREATE MATERIALIZED VIEW tab_mv TO tab AS
SELECT ... sendAddress, ...
FROM tab_kfk;

然后我尝试从第一个表中选择所有或特定项目 - 选项卡，但什么也得不到。日志关注

好的。只需在 sendAddress 中的花括号前添加 '[]'，如下所示：

"authkey":"some_value",
   "sendAddress":[{
      "sendCommChannelTypeId":4,
      "sendCommChannelTypeCode":"SMS",
      "sendAddress":"789345345945"
   }]

我仍然有一个错误，但略有不同：我该怎么做才能解决这个问题，谢谢！

【问题讨论】：

标签： apache-kafka clickhouse

【解决方案1】：

有 3 种方法可以解决它：

在插入 Kafka 主题之前不要使用嵌套对象和展平消息。比如这样的方式：

{
    ..
    "authkey":"key",
    "sendAddress_CommChannelTypeId":4,
    "sendAddress_CommChannelTypeCode":"SMS",
    "sendAddress":"789345345945",
    ..
}

使用更改 JSON 消息架构和表架构所需的 Nested data structure：

{
    ..
    "authkey":"key",
    "sendAddress.sendCommChannelTypeId":[4],
    "sendAddress.sendCommChannelTypeCode":["SMS"],
    "sendAddress.sendAddress":["789345345945"],
    ..
}

CREATE TABLE tab_kfk
(
    applicationId Int32,
    ..
    sendAddress Nested(
        sendCommChannelTypeId Int32,
        sendCommChannelTypeCode String,
        sendAddress String),
    ..
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'localhost:9092',
       kafka_topic_list = 'topk2',
       kafka_group_name = 'group1',
       kafka_format = 'JSONEachRow',
       kafka_row_delimiter = '\n',
       input_format_import_nested_json = 1 /* <--- */

考虑设置input_format_import_nested_json。

将输入的 JSON 消息解释为字符串并手动解析（参见 github issue #16969）：

CREATE TABLE tab_kfk
(
    message String
)
ENGINE = Kafka
SETTINGS 
    ..
    kafka_format = 'JSONAsString', /* <--- */
    ..

CREATE MATERIALIZED VIEW tab_mv TO tab 
AS
SELECT 
    ..
    JSONExtractString(message, 'authkey') AS authkey,
    JSONExtract(message, 'sendAddress', 'Tuple(Int32,String,String)') AS sendAddress,
    ..
FROM tab_kfk;

【讨论】：