说明

Table 对象可以转 View 视图，关键代码：

Table table = stEnv.fromDataStream(soso);
// table 转 view
stEnv.createTemporaryView("v_order",table);

例子

package p2socket;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class C5table转view {
    public static void main(String[] args) throws Exception {
        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.fromElements(
                "java,50,lilei",
                "hadoop,90,lilei",
                "flink,70,lilei"
                );

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

        Table table = stEnv.fromDataStream(soso);
        // table 转 view
        stEnv.createTemporaryView("v_order",table);
        Table res=stEnv.sqlQuery("select name,price from v_order where price>60");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();
        /*
        将查询结果，封装到 Row 类对象中
         */
        DataStream<Row> dsRes = stEnv.toAppendStream(res, Row.class);
        dsRes.print();
        env.execute();
    }
}

原文出处：/show_1IX4E6BPVlxt.html