如何用嵌套字段注册Flink表模式？

Question

我正在StreamingTableEnvironment中将数据流注册为Flink表。

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

tableEnv.createTemporaryView(
       "foo",
       dataStream,
       $("f0").as("foo-id"),
       $("f1").as("foo-value") 
)

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'
);

我没有使用Create Table命令with connector注册流，因为数据格式是自定义的，这就是我为什么要注册流的原因。

Answer 1

我尝试了一些返回类型和嵌套，这个解决方案似乎很管用：

DataStream<Row> testDataStream = env.fromCollection(Arrays.asList(
     Row.of("sherin", 1L, Row.of(100L, 200L)),
     Row.of("thomas", 1L, Row.of(100L, 200L))
)).returns(
    Types.ROW(
             Types.STRING,
             Types.LONG,
             Types.ROW_NAMED(
                  new String[]{"val1", "val2"},
                  Types.LONG,
                  Types.LONG)
));

tableEnv.createTemporaryView(
    "foo",
    testDataStream,
    $("f0").as("name"),
    $("f1").as("c"),
    $("f2").as("age"));

Table testTable = tableEnv.sqlQuery(
     "select name, c, age.val1, age.val2 from foo"
);

DataStream<Row> result = tableEnv.toAppendStream(
    testTable,
    TypeInformation.of(Row.class)
);

result.print().setParallelism(2);

如果有更好的方法的话，我仍然愿意接受各种想法。