189 8069 5689

如何使用TbaleSQL与FlinkJDBC连接器读取MYSQL数据

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

创新互联始终坚持【策划先行，效果至上】的经营理念，通过多达10年累计超上千家客户的网站建设总结了一套系统有效的网络营销推广解决方案，现已广泛运用于各行各业的客户，其中包括：成都茶楼设计等企业，备受客户称赞。

使用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 out) throws Exception {
                if (value.f0) {
                    System.out.println(value.f1.toString());
                }
            }
        }).print();

        env.execute();
    }

    /**
     * avg 获取一组数据流中的数值平均值
     * @param tEnv
     * @return
     */
    public static Table avg(StreamTableEnvironment tEnv){
        //第一种：执行SQL
        String sql = "select status,avg(age) as age1 from my_users group by status";
        //Table table = tEnv.sqlQuery(sql);

        //第二种：通过方法拼装执行语句
        Table table = tEnv.from("my_users").groupBy($("status")).select($("status"),$("age").avg().as("age1"));
        return table;
    }

    /**
     * count 获取一组数据流中累加分组的行数之和
     * @param tEnv
     * @return
     */
    public static Table count(StreamTableEnvironment tEnv){
        //第一种：执行SQL
        String sql = "select status,count(age) as age1 from my_users group by status";
        //Table table = tEnv.sqlQuery(sql);

        //第二种：通过方法拼装执行语句
        Table table = tEnv.from("my_users").groupBy($("status")).select($("status"),$("age").count().as("age1"));
        return table;
    }

    /**
     * sum 获取一组数据流中累加分组的数值之和
     * @param tEnv
     * @return
     */
    public static Table sum(StreamTableEnvironment tEnv){
        //第一种：执行SQL
        String sql = "select status,sum(age) as age1 from my_users group by status";
        //Table table = tEnv.sqlQuery(sql);

        //第二种：通过方法拼装执行语句
        Table table = tEnv.from("my_users").groupBy($("status")).select($("status"),$("age").sum().as("age1"));
        return table;
    }


    /**
     * min 获取一组数据流中的最小值
     * @param tEnv
     * @return
     */
    public static Table min(StreamTableEnvironment tEnv){
        //第一种：执行SQL
        String sql = "select status,min(age) as age1 from my_users group by status";
        //Table table = tEnv.sqlQuery(sql);

        //第二种：通过方法拼装执行语句
        Table table = tEnv.from("my_users").groupBy($("status")).select($("status"),$("age").min().as("age1"));
        return table;
    }

    /**
     * max 获取一组数据流中的最大值,每一次数据加入，则计算一次
     * @param tEnv
     * @return
     */
    public static Table max(StreamTableEnvironment tEnv){
        //第一种：执行SQL
        String sql = "select status,max(age) as age1 from my_users group by status";
        //Table table = tEnv.sqlQuery(sql);

        //第二种：通过方法拼装执行语句
        Table table = tEnv.from("my_users").groupBy($("status")).select($("status"),$("age").max().as("age1"));
        return table;
    }

}
建表SQL
CREATE TABLE `users` (
  `id` bigint(8) NOT NULL AUTO_INCREMENT,
  `name` varchar(40) DEFAULT NULL,
  `age` int(8) DEFAULT NULL,
  `status` tinyint(2) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
打印结果
root
 |-- status: INT
 |-- age1: INT

0,16
0,18
1,21
1,28
2,31
感谢各位的阅读，以上就是“如何使用Tbale SQL与Flink JDBC连接器读取MYSQL数据”的内容了，经过本文的学习后，相信大家对如何使用Tbale SQL与Flink JDBC连接器读取MYSQL数据这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！
            
            
                        

            文章名称：如何使用TbaleSQL与FlinkJDBC连接器读取MYSQL数据            

            标题URL：http://cdxtjz.cn/article/jcodeh.html
        
    
    

        
其他资讯

        
    




  

    

      
联系我们

      

        您好HELLO!

        感谢您来到成都网站建设公司，若您有合作意向，请您为我们留言或使用以下方式联系我们，

        我们将尽快给你回复，并为您提供真诚的设计服务，谢谢。
      

      
    
        
  • 电话：028- 86922220 18980695689
    • 
        
  • 商务合作邮箱：631063699@qq.com
    • 
        
  • 合作QQ： 532337155
    • 
        
  • 成都网站设计地址：成都市青羊区锣锅巷31号五金站写字楼6楼
    • 
      

    

    

      
小谭建站工作室

      

        成都小谭网站建设公司拥有多年以上互联网从业经验的团队，始终保持务实的风格，以"帮助客户成功"为已任，专注于提供对客户有价值的服务。

        我们已为众企业及上市公司提供专业的网站建设服务。我们不只是一家网站建设的网络公司；我们对营销、技术、管理都有自己独特见解，小谭建站采取“创意+综合+营销”一体化的方式为您提供更专业的服务！
      

      
小谭观点

      

        相对传统的成都网站建设公司而言，小谭是互联网中的网站品牌策划，我们精于企业品牌与互联网相结合的整体战略服务。

        我们始终认为，网站必须注入企业基因，真正使网站成为企业vi的一部分，让整个网站品牌策划体系变的深入而持久。
      

    

  

  

    
 
         友情链接：
        成都木托盘   网站设计   西云主机托管   服务器租凭   重庆app软件开发   广告品牌策划设计   成都400电话   成都西信服务器托管   化妆品企业网站建设方案   无边发光字     
    

  

  
 © Copyright 2023 小谭建站工作室All Rights Reserved.  成都网站建设 / 成都网站建设 / 响应式网站建设 / 成都服务器租用