使用 Apache Flink SQL 从 Kafka 消息中获取嵌套字段答案

【问题标题】：Get nested fields from Kafka message using Apache Flink SQL使用 Apache Flink SQL 从 Kafka 消息中获取嵌套字段
【发布时间】：2021-01-09 11:10:58
【问题描述】：

我正在尝试使用 Apache Flink 1.11 创建一个源表，我可以在其中访问 JSON 消息中的嵌套属性。我可以从根属性中提取值，但我不确定如何访问嵌套对象。

documentation 建议它应该是 MAP 类型，但是当我设置它时，我收到以下错误

: java.lang.UnsupportedOperationException: class org.apache.calcite.sql.SqlIdentifier: MAP

这是我的 SQL

        CREATE TABLE input(
            id VARCHAR,
            title VARCHAR,
            properties MAP
        ) WITH (
            'connector' = 'kafka-0.11',
            'topic' = 'my-topic',
            'properties.bootstrap.servers' = 'localhost:9092',
            'properties.group.id' = 'python-test',
            'format' = 'json'
        )

我的 JSON 看起来像这样：

{
  "id": "message-1",
  "title": "Some Title",
  "properties": {
    "foo": "bar"
  }
}

【问题讨论】：

标签： apache-flink flink-sql pyflink

【解决方案1】：

您可以使用ROW 提取 JSON 消息中的嵌套字段。您的 DDL 语句如下所示：

CREATE TABLE input(
             id VARCHAR,
             title VARCHAR,
             properties ROW(`foo` VARCHAR)
        ) WITH (
            'connector' = 'kafka-0.11',
            'topic' = 'my-topic',
            'properties.bootstrap.servers' = 'localhost:9092',
            'properties.group.id' = 'python-test',
            'format' = 'json'
        );

【讨论】：

太棒了，这行得通！深度嵌套的 JSON 怎么样，看起来这可能会变得笨拙？使用 PyFlink 和 SQL 处理 JSON 数据有没有更好的方法？
目前没有更好的方法可以做到这一点，AFAIK。不过，我们计划很快在 Flink SQL 中支持 SQL JSON 函数：cwiki.apache.org/confluence/pages/… 这应该会让事情变得容易得多！

【解决方案2】：

你也可以试试

CREATE TABLE input(
            id VARCHAR,
            title VARCHAR,
            properties MAP<STRING, STRING>
        ) WITH (
            'connector' = 'kafka-0.11',
            'topic' = 'my-topic',
            'properties.bootstrap.servers' = 'localhost:9092',
            'properties.group.id' = 'python-test',
            'format' = 'json'
        )

唯一的区别是：MAP<STRING, STRING> 与 MAP

【讨论】：