Flink：数据流到表

Question

不幸的是，Kafka Flink连接器只支持-csv、json和avro格式。因此，我不得不使用较低级别的API（数据流）。

问题：如果我可以从datastream对象中创建一个表，那么我就可以接受在该表上运行的查询。它将使转换部分无缝和通用。是否可以在数据流对象上运行SQL查询？

Answer 1

如果您有一个Datastream对象，那么您可以简单地使用StreamTableEnvironment将给定的Datastream注册为表。

这将看起来或多或少如下所示：

val myStream = ...
val env: StreamExecutionEnvironment = configureFlinkEnv(StreamExecutionEnvironment.getExecutionEnvironment)
val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
tEnv.registerDataStream("myTable", myStream, [Field expressions])

那么您应该能够查询从您的数据流创建的动态表。