如何使用Tbale SQL与Flink JDBC连接器读取MYSQL数据

这篇文章主要讲解了"如何使用Tbale SQL与Flink JDBC连接器读取MYSQL数据"，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习"如何使用Tbale SQL与Flink JDBC连接器读取MYSQL数据"吧！

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

示例环境

java.version: 1.8.xflink.version: 1.11.1kafka：2.11

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.$;/** * @Description 使用Tbale&SQL与Flink JDBC连接器读取MYSQL数据，并用GROUP BY语句根据一个或多个列对结果集进行分组。 */public class GroupToMysql {    /**     官方参考：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html     分区扫描     为了加速并行Source任务实例中的数据读取，Flink为JDBC表提供了分区扫描功能。     scan.partition.column：用于对输入进行分区的列名。     scan.partition.num：分区数。     scan.partition.lower-bound：第一个分区的最小值。     scan.partition.upper-bound：最后一个分区的最大值。     */    //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 ENFORCED\n" +                    ") 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