Flink 系例 之 TableAPI & SQL 与 MYSQL 分组统计

使用 Tbale&SQL 与 Flink JDBC 连接器读取 MYSQL 数据，并用 GROUP BY 语句根据一个或多个列对结果集进行分组。

示例环境

java.version: 1.8.x
flink.version: 1.11.1
kafka：2.11

示例数据源 （项目码云下载）

Flink 系例 之 搭建开发环境与数据

示例模块 （pom.xml）

Flink 系例 之 TableAPI & SQL 与 示例模块

GroupToMysql.java

package com.flink.examples.mysql;

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.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 GroupToMysql {

    

    //flink-jdbc-1.11.1写法,所有属性名在JdbcTableSourceSinkFactory工厂类中定义
    static String table_sql =
            "CREATE TABLE my_users (n" +
                    "  id BIGINT,n" +
                    "  name STRING,n" +
                    "  age INT,n" +
                    "  status INT,n" +
                    "  PRIMARY KEY (id) NOT ENFORCEDn" +
                    ") WITH (n" +
                    "  'connector.type' = 'jdbc',n" +
                    "  'connector.url' = 'jdbc:mysql://192.168.110.35:3306/flink?useUnicode=true&characterEncoding=utf-8', n" +
                    "  'connector.driver' = 'com.mysql.jdbc.Driver', n" +
                    "  'connector.table' = 'users', n" +
                    "  'connector.username' = 'root',n" +
                    "  'connector.password' = 'password' n" +
//                    "  'connector.read.fetch-size' = '10' n" +
                    ")";

    public static void main(String[] args) throws Exception {
        //构建StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置setParallelism并行度
        env.setParallelism(1);
        //构建EnvironmentSettings 并指定Blink Planner
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        //构建StreamTableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
        //注册mysql数据维表
        tEnv.executeSql(table_sql);

        //Table table = avg(tEnv);
        //Table table = count(tEnv);
        //Table table = min(tEnv);
        Table table = max(tEnv);

        //打印字段结构
        table.printSchema();

        //普通查询操作用toAppendStream
        //tEnv.toAppendStream(table, Row.class).print();
        //group操作用toRetractStream
        //tEnv.toRetractStream(table, Row.class).print();

        //table 转成 dataStream 流，Tuple2第一个参数flag是true表示add添加新的记录流，false表示retract表示旧的记录流
        DataStream> behaviorStream = tEnv.toRetractStream(table, Row.class);
        behaviorStream.flatMap(new FlatMapFunction, Object>() {
            @Override
            public void flatMap(Tuple2 value, Collector