flink sql

【案例1】Flink01_Table_BaseUse

public class Flink01_Table_BaseUse {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 1),
                new WaterSensor("sensor_2", 2000L, 2),
                new WaterSensor("sensor_1", 3000L, 3),
                new WaterSensor("sensor_1", 4000L, 4),
                new WaterSensor("sensor_2", 5000L, 5),
                new WaterSensor("sensor_1", 6000L, 6)
        );

        //1.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //2.通过表的执行环境，将流转成表（动态表）
        Table table = tEnv.fromDataStream(stream);
        //table.printSchema();   //查看动态表的 元数据

        //3.在动态表上执行连续查询
        // select id,ts,vc from t where id='sensor_1'
        Table result = table.where("id='sensor_1'").select("id,ts,vc");

        //4.把动态表转成流
        DataStream<WaterSensor> resultStream = tEnv.toAppendStream(result, WaterSensor.class);
        resultStream.print();
        
        env.execute();
        
    }
}

运行结果:

WaterSensor(id=sensor_1, ts=1000, vc=1)
WaterSensor(id=sensor_1, ts=3000, vc=3)
WaterSensor(id=sensor_1, ts=4000, vc=4)
WaterSensor(id=sensor_1, ts=6000, vc=6)

【案例1.2】Flink01_Table_BaseUse2

public class Flink01_Table_BaseUse {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 1),
                new WaterSensor("sensor_2", 2000L, 2),
                new WaterSensor("sensor_1", 3000L, 3),
                new WaterSensor("sensor_1", 4000L, 4),
                new WaterSensor("sensor_2", 5000L, 5),
                new WaterSensor("sensor_1", 6000L, 6)
        );

        //1.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //2.通过表的执行环境，将流转成表（动态表）
        Table table = tEnv.fromDataStream(stream);
        //table.printSchema();   //查看动态表的 元数据

        //3.在动态表上执行连续查询
        //  select id,ts,vc from t where id='sensor_1'

        Table result = table.where("id='sensor_1'").select("id,vc");

        //4.把动态表转成流
        DataStream<Row> resultStream = tEnv.toAppendStream(result, Row.class);
        resultStream.print();

        env.execute();
    }
}

运行结果：

+I[sensor_1, 1]
+I[sensor_1, 3]
+I[sensor_1, 4]
+I[sensor_1, 6]