说明

文中案例，连接 socket ，从socket读取数据，实现分组聚合查询

在 flink1.12.x SQL：创建工程，从socket读取数据，实现查询 基础上实现

java

主类

package p2socket;

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 C3Socket分组查询 {
    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
        stEnv.createTemporaryView("t_order",soso,$("name"),$("price"),$("username"));
        Table res=stEnv.sqlQuery("select username,avg(price) from t_order where price>60 group by username");
        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();
    }
}

注意：

• 临时视图，不要出现关键字，如：order 等

• 因为查询列与封装类不同，所以改用 Row 封装

测试

启动 socket 服务器

略

启动 flink 程序

执行结果：

Schema信息:
root
 |-- name: STRING
 |-- price: DOUBLE
 |-- username: STRING

发送第一个数据

发送下面数据

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_1IX4E2bjQrdy.html