KSQL - 从结构数组中选择列作为数组答案

【问题标题】：KSQL - Select Columns from Array of Struct as ArraysKSQL - 从结构数组中选择列作为数组
【发布时间】：2020-06-24 21:55:48
【问题描述】：

类似于KSQL streams - Get data from Array of Struct，我的输入 JSON 看起来像：

{
  "Obj1": {
    "a": "abc",
    "b": "def",
    "c": "ghi"
  },
  "ArrayObj": [
    {
      "key1": "1",
      "key2": "2",
      "key3": "3"
    },
    {
      "key1": "4",
      "key2": "5",
      "key3": "6"
    },
    {
      "key1": "7",
      "key2": "8",
      "key3": "9"
    }
  ]
}

我创建了一个流：

CREATE STREAM Example1(Obj1 STRUCT<a VARCHAR, b VARCHAR, c VARCHAR>, ArrayObj ARRAY<STRUCT<key1 VARCHAR, key2 VARCHAR, key3 VARCHAR>>) WITH (kafka_topic='sample_topic', value_format='JSON', partitions=1);

但是，我希望每个输入 JSON 文档只有一行输出，数组中每一列的数据都被展平为数组，例如：

 a    b   key1      key2      key3

 abc  def [1, 4, 7] [2, 5, 8] [3, 6, 9]

这可以用 KSQL 实现吗？

【问题讨论】：

标签： ksqldb

【解决方案1】：

目前你只能以你想要的方式展平ArrayObj，前提是你预先知道它将有多少个元素：

CREATE STREAM flatten AS
  SELECT
    Obj1.a AS a,
    Obj1.b AS b,
    ARRAY[ArrayObj[1]['key1'], ArrayObj[2]['key1'], ArrayObj[3]['key1']] as key1,
    ARRAY[ArrayObj[1]['key2'], ArrayObj[2]['key2'], ArrayObj[3]['key2']] as key2,
    ARRAY[ArrayObj[1]['key3'], ArrayObj[2]['key3'], ArrayObj[3]['key3']] as key3,
  FROM Example1;

我猜如果你新的数组要达到一定的大小，你可以只用一个case语句来选择性地提取元素，例如

-- handles arrays of size 2 or 3 elements, i.e. third element is optional.
CREATE STREAM flatten AS
  SELECT
    Obj1.a AS a,
    Obj1.b AS b,
    ARRAY[ArrayObj[1]['key1'], ArrayObj[2]['key1'], ArrayObj[3]['key1']] as key1,
    ARRAY[ArrayObj[1]['key2'], ArrayObj[2]['key2'], ArrayObj[3]['key2']] as key2,
    CASE
      WHEN ARRAY_LENGTH(ArrayObj) >= 3)
        THEN ARRAY[ArrayObj[1]['key3'], ArrayObj[2]['key3'], ArrayObj[3]['key3']] 
      ELSE
        null
      as key3,
  FROM Example1;

如果这不符合您的需求，那么目前围绕 ksqlDB 中的 lambda 函数支持进行的设计讨论可能会引起您的兴趣：https://github.com/confluentinc/ksql/pull/5661

【讨论】：