可以使用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'
        );

[2022年更新]

在Apache Flink 1.13版本中没有系统内置JSON函数。它们在1.14版本中引入。检查this
如果您使用的是版本〈1.14，请参阅下面的解决方案。

如何使用嵌套JSON输入创建表？

JSON输入示例：

{
    "id": "message-1",
    "title": "Some Title",
    "properties": {
    "foo": "bar"
    "nested_foo":{
        "prop1" : "value1",
        "prop2" : "value2"
    }
}

创建语句

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

如何选择嵌套列？

SELECT properties.foo, properties.nested_foo.prop1 FROM input;

请注意，如果使用输出结果

SELECT properties FROM input

您将看到行格式的结果。列properties的内容将是

+I[bar, +I[prop1,prop2]]

你也可以试试

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

如果使用format=raw，您可以使用JSON_VALUE函数从payload中提取感兴趣的字段：下面是代码：

CREATE TABLE input(
        payload STRING,
        foo AS JSON_VALUE(payload, '$.properties.foo' RETURNING STRING),
) WITH (
  'connector' = 'kafka',        
  'format' = 'raw'
)