创建工程参见：flink1.12.x Table API 和 SQL：创建工程，读取csv文件，实现查询（静态表）

介绍

实现分组聚合查询，sql如下：

select username,avg(price) from t_order where price>60 group by username

由于求 price 的平均数，即：每来一条数据，该查询结果就发生变化，所以不能用 Append 追加，而应该用 Retract 对结果进行更新

java

package test;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


public class TestTable读取文件_分组sql {
    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);

        /*
        注意不要出现关键字，如：order等
        需要依赖flink-csv
         */
        TableResult table=stEnv.executeSql("CREATE TABLE t_order (" +
                "  name STRING," +
                "  price double," +
                "  username STRING" +
                ") WITH ( " +
                "  'connector' = 'filesystem'," +
                "  'path' = 'file:///D:/bigdata/flink_table_sql/data'," +
                "  'format' = 'csv'," +
                "  'csv.field-delimiter' = ','" +
                ")");
        //执行查询
        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，才能打印结果，需要指定 封装的泛型
        将计算后的数据，在ds原数据基础上更新
        true：表示更新
        false：表示删除
         */
        DataStream<Tuple2<Boolean, Row>> dsRes = stEnv.toRetractStream(res, Row.class);
        dsRes.print();
        /*
        将计算后的数据，append到ds，支持简单查询
        分组查询，来一条记录，需要重新计算，并更新原结果，所以 toAppendStream 方法不支持
         */
//        DataStream<Row> dsRes2 = stEnv.toAppendStream(res, Row.class);
//        dsRes2.print();

        env.execute();
    }

}

执行结果

Schema信息:
root
 |-- username: STRING
 |-- EXPR$1: DOUBLE

2> (true,lucy,120.0)
4> (true,韩梅梅,90.0)
2> (true,李雷,80.0)
2> (false,李雷,80.0)
2> (true,李雷,90.0)
4> (false,韩梅梅,90.0)
4> (true,韩梅梅,80.0)
2> (false,lucy,120.0)
2> (true,lucy,110.0)

来一条记录，需要 重新计算，并 更新原结果，上面结果中，false 表示删除，true 表示新增

注意

分组查询，来一条记录，需要重新计算，并更新原结果，所以 toAppendStream 方法会报错

原文出处：/show_1IX23ORfmWjZ.html