说明
为了使用关系查询处理流，必须将其转换成 Table。
基于一个 DataStream，先流式 地读取数据源，然后 map 成 POJO，再把它转成 Table （或 View）。
当流中有新数据到来，动态表中会插入一行记录
字段
Table 的列字段（column fields），就是 POJO 里的字段，这样就不用再麻烦地定义 schema 了。
关键代码
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
//构建Table环境
EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);
// 流
DataStreamSource<String> dss = env.socketTextStream("127.0.0.1", 9999, "\n");
// map操作，将流中数据 转成 pojo
SingleOutputStreamOperator<Order> soso = dss.flatMap(new FlatMapFunction<String, Order>() {
@Override
public void flatMap(String value, Collector<Order> out) throws Exception {
String[] arr = value.split(",");
Double price = Double.parseDouble(arr[1]);
Order order = new Order(arr[0], price, arr[2]);
out.collect(order);
}
});
// 创建临时视图，不要使用关键字，如：order
stEnv.createTemporaryView("t_order",soso);
参考：
https://blog.csdn.net/Suyebiubiu/article/details/114688317