如何从kafka的主题中创建具有大量JSON字段的KSQL Stream？

Question

我将一个长 JSON 字符串传递给 kafka 主题，例如：

{
    "glossary": {
        "title": "example glossary",
        "GlossDiv": {
            "title": "S",
            "GlossList": {
                "GlossEntry": {
                    "ID": "SGML",
                    "SortAs": "SGML",
                    "GlossTerm": "Standard Generalized Markup Language",
                    "Acronym": "SGML",
                    "Abbrev": "ISO 8879:1986",
                    "GlossDef": {
                        "para": "A meta-markup language, used to create markup languages such as DocBook.",
                        "GlossSeeAlso": ["GML", "XML"]
                    },
                    "GlossSee": "markup"
                }
            }
        }
    }
}

并希望从 kafka 主题创建包含所有字段的流，而不指定 KSQL 中的每个字段，例如：

 CREATE STREAM pageviews_original (*) WITH \
(kafka_topic='pageviews', value_format='JSON');

Answer 1

如果您希望 KSQL 自动提取字段名称，则需要使用 Avro。如果您使用 Avro，则数据的架构会在 Confluent Schema Registry 中注册，并且 KSQL 会在您使用主题时自动检索它。

如果您使用 JSON，您必须告诉 KSQL 列是什么。您可以在CREATE STREAM 语句中执行此操作，对嵌套元素使用STRUCT 数据类型。

您可以通过仅声明CREATE STREAM 中的高级字段然后使用EXTRACTJSONFIELD 访问您要使用的字段的嵌套元素来列出所有字段。请注意，5.0.0 中存在一个问题，即fixed in 5.0.1。此外，您不能将其用于您显示的示例数据中的嵌套数组等。