KSQL 流 - 遍历 JSON答案

【问题标题】：KSQL streams - Iterate over a JSONKSQL 流 - 遍历 JSON
【发布时间】：2020-01-11 09:21:13
【问题描述】：

如何在 KSQL 流中迭代 JSON：

我的 JSON 看起来像：

{
   "Obj" {
       "ID" : "1"
    },
    "KeyValues": {
        "Key1": "value1",
        "Key2": "value2",
        "Key3": "value3",
        "Key4": "value4",
        "Key5": "value5",
        "Key6": "value6",
        "Key7": "value7",
        "Key8": "value8",
        "Key9": "value9",
        "Key10": "value10",
            |
            |
            |
            |

        "KeyN": "valueN"
  }
}

如何在 KSQL 中低于 o/p。需要为 N 个元素遍历 JSON 对象并列出如下。

   ID    KEY              VALUE
----------------------------------    
   1     Key1            value1
   1     Key2            value2
   1     Key3            value3
   1     Key4            value4
   1     Key5            value5
   1     Key6            value6
   1     Key7            value7
   1     Key8            value8
   1     Key9            value9
   1     Key10           value10
   1            |
   1            |
   1            |
   1            |
   1 
   1      KeyN          valueN

提前致谢。

【问题讨论】：

如果是“N”长度，KeyValues 不应该是一个数组吗？
@RobinMoffatt 这不是一个数组。通常它应该是数组，但这个要求与 JSON 一样，即没有数组的 JOSN 对象。
我也有同样的场景。它不是一个数组。在ksql中有什么可能吗？
迭代时具体需要做什么？也许你可以写一个UDF？
docs.confluent.io/current/ksql/docs/developer-guide/…

标签： apache-kafka ksqldb confluent-platform

【解决方案1】：

似乎UDTF 是适合您的解决方案。您可以将explode 视为一个 UDTF 示例，它接收一个数组，然后输出 N 行，每个元素一个。

您的 UDTF 的签名将是相似的：

@Udtf(schema = "STRUCT<key VARCHAR, value VARCHAR>")
public <T> List<Struct> expandMapEntries(final Map<String, String> input) {
  // output a list of key value pairs as a struct from 'input'
}

然后您可以使用此 UDTF 并从中选择字段（如下所示）：

CREATE STREAM expanded AS SELECT EXPAND_MAP_ENTRIES(KeyValues) AS keyVals FROM source;
CREATE STREAM flattened AS keyVals->key as `KEY`, keyVals->value AS VALUE FROM expanded;

如果这对您有用，请告诉我，并随时联系社区 slack (@almog) - 我对这个用例非常感兴趣。

【讨论】：