apache-kafka - 如何使用kafka主题中的大量JSON字段创建KSQL流？

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，数据的架构会在 Confluence Schema Registry 中注册，当您使用主题时，KSQL 将自动检索它。

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

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