说明

文中案例，连接 socket ，从socket读取数据，使用 table api 实现查询

工程结构、依赖、配置与 flink1.12.x SQL：创建工程，从socket读取数据，实现查询 相同

java

封装类

package p2socket;

public class Order {
    private String name;
    private Double price;
    private String username;

    // 要有无参构造方法
    public Order() {
    }

    public Order(String name, Double price, String username) {
        this.name = name;
        this.price = price;
        this.username = username;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Double getPrice() {
        return price;
    }

    public void setPrice(Double price) {
        this.price = price;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    @Override
    public String toString() {
        return "Order{" +
                "name='" + name + '\'' +
                ", price=" + price +
                ", username='" + username + '\'' +
                '}';
    }
}

注意：要有无参构造方法

主类

package p3tableapi;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
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;

import static org.apache.flink.table.api.Expressions.$;

public class C1SocketTable {
    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.socketTextStream("127.0.0.1", 9999, "\n");
        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
        Table table = stEnv.fromDataStream(soso, $("name"), $("price"), $("username"));

        // table api
        Table res = table
                .where($("price").isGreater(60)) // 先设置 where过滤
                .groupBy($("username")) // 在设置分组
                // 最后指定select查询字段、聚合函数，设置别名
                .select($("username"), $("price").avg().as("avg_price"));

        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();
        /*
        需要将结果转换成DataStream，才能打印结果，指定 封装的泛型 Row，因为查询列与封装类不同，所以改用 Row封装
        将计算后的数据，在ds原数据基础上更新
        true：表示更新
        false：表示删除
         */
        DataStream<Tuple2<Boolean, Row>> dsRes = stEnv.toRetractStream(res, Row.class);
        dsRes.print();
        env.execute();
    }
}

测试

启动 socket 服务器

略

启动 flink 程序

执行结果：

Schema信息:
root
 |-- username: STRING
 |-- avg_price: DOUBLE    //别名

发送第一个数据

发送下面数据

java,50,lilei

执行结果：没有数据打印，因为价格低于 60

发送第二个数据

hadoop,90,lilei

执行结果：

4> (true,lilei,90.0) // true，表示添加数据

发送第三个数据

flink,70,lilei

执行结果

4> (false,lilei,90.0) // false，表示将上一条 lilei,90.0 数据删除
4> (true,lilei,80.0)// true，表示添加数据：lilei,80.0，平均值是 80

原文出处：/show_1IX4E568GPn5.html