您好，登录后才能下订单哦！
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
这篇文章主要讲解了“如何使用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<Tuple2<Boolean, Row>> behaviorStream = tEnv.toRetractStream(table, Row.class);
behaviorStream.flatMap(new FlatMapFunction<Tuple2<Boolean, Row>, Object>() {
@Override
public void flatMap(Tuple2<Boolean, Row> value, Collector<Object> 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数据这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。