189 8069 5689
这篇文章主要讲解了“Flink Connectors怎么连接MySQL”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Flink Connectors怎么连接MySql”吧！
成都创新互联公司主要从事成都网站设计、成都网站制作、网页设计、企业做网站、公司建网站等业务。立足成都服务开福,十年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:13518219792
通过使用Flink DataStream Connectors 数据流连接器连接到Mysql数据源，并基于JDBC提供数据流输入与输出操作
示例环境
java.version: 1.8.xflink.version: 1.11.1mysql：5.7.x
数据流输入
DataStreamSource.java
package com.flink.examples.mysql;
import com.flink.examples.TUser;
import com.google.gson.Gson;
import org.apache.flink.api.java.io.jdbc.JDBCOptions;
import org.apache.flink.api.java.io.jdbc.JDBCTableSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/**
* @Description 将mysql表中数据查询输出到DataStream流中
*/
public class DataStreamSource {
/**
* 官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/jdbc.html
*/
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//查询sql
String sql = "SELECT id,name,age,sex,address,createTimeSeries FROM t_user";
//设置表视图字段与类型
TableSchema tableSchema = TableSchema.builder()
.field("id", DataTypes.INT())
.field("name", DataTypes.STRING())
.field("age", DataTypes.INT())
.field("sex", DataTypes.INT())
.field("address", DataTypes.STRING())
//.field("createTime", DataTypes.TIMESTAMP())
.field("createTimeSeries", DataTypes.BIGINT())
.build();
//配置jdbc数据源选项
JDBCOptions jdbcOptions = JDBCOptions.builder()
.setDriverName(MysqlConfig.DRIVER_CLASS)
.setDBUrl(MysqlConfig.SOURCE_DRIVER_URL)
.setUsername(MysqlConfig.SOURCE_USER)
.setPassword(MysqlConfig.SOURCE_PASSWORD)
.setTableName("t_user")
.build();
JDBCTableSource jdbcTableSource = JDBCTableSource.builder().setOptions(jdbcOptions).setSchema(tableSchema).build();
//将数据源注册到tableEnv视图student中
tEnv.registerTableSource("t_user", jdbcTableSource);
Table table = tEnv.sqlQuery(sql);
DataStream sourceStream = tEnv.toAppendStream(table, TUser.class);
sourceStream.map((t)->new Gson().toJson(t)).print();
env.execute("flink mysql source");
}
} 数据流输出
DataStreamSink.java
package com.flink.examples.mysql;
import com.flink.examples.TUser;
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.sinks.TableSink;
import static org.apache.flink.table.api.Expressions.$;
/**
* @Description 将DataStream数据流插入到mysql表中
*/
public class DataStreamSink {
/**
* 官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/jdbc.html
*/
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.enableCheckpointing(2000);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//查询sql
String sql = "insert into t_user (id,name,age,sex,address,createTimeSeries) values (?,?,?,?,?,?)";
//封装数据
TUser user = new TUser();
user.setId(0);
user.setName("zhao1");
user.setAge(22);
user.setSex(1);
user.setAddress("CN");
user.setCreateTimeSeries(System.currentTimeMillis());
DataStream sourceStream = env.fromElements(user);
//从DataStream获取数据
// Expression id = ExpressionParser.parseExpression("id");
// Expression name = ExpressionParser.parseExpression("name");
// Expression age = ExpressionParser.parseExpression("age");
// Expression sex = ExpressionParser.parseExpression("sex");
// Expression address = ExpressionParser.parseExpression("address");
// Expression createTimeSeries = ExpressionParser.parseExpression("createTimeSeries");
// Table table = tEnv.fromDataStream(sourceStream, id, name, age, sex, address, createTimeSeries );
Table table = tEnv.fromDataStream(sourceStream,$("id"),$("name"),$("age"),$("sex"),$("address"),$("createTimeSeries"));
//输出到mysql
//设置表视图字段与类型
TableSchema tableSchema = TableSchema.builder()
.field("id", DataTypes.INT())
.field("name", DataTypes.STRING())
.field("age", DataTypes.INT())
.field("sex", DataTypes.INT())
.field("address", DataTypes.STRING())
//.field("createTime", DataTypes.TIMESTAMP())
.field("createTimeSeries", DataTypes.BIGINT())
.build();
//设置sink输出jdbc
TableSink tableSink = JDBCAppendTableSink.builder()
.setDrivername(MysqlConfig.DRIVER_CLASS)
.setDBUrl(MysqlConfig.SOURCE_DRIVER_URL)
.setUsername(MysqlConfig.SOURCE_USER)
.setPassword(MysqlConfig.SOURCE_PASSWORD)
.setQuery(sql)
.setParameterTypes(tableSchema.getFieldTypes())
.setBatchSize(100)
.build();
//将数据源注册到tableEnv视图result中
tEnv.registerTableSink("result",
tableSchema.getFieldNames(),
tableSchema.getFieldTypes(),
tableSink);
//在指定的路径下注册，然后执行插入操作
table.executeInsert("result");
}
} 数据源配置类
MysqlConfig.java
package com.flink.examples.mysql;
/**
* @Description Mysql数据库连接配置
*/
public class MysqlConfig {
public final static String DRIVER_CLASS="com.mysql.jdbc.Driver";
public final static String SOURCE_DRIVER_URL="jdbc:mysql://127.0.0.1:3306/flink?useUnicode=true&characterEncoding=utf-8&useSSL=false";
public final static String SOURCE_USER="root";
public final static String SOURCE_PASSWORD="root";
}数据展示
感谢各位的阅读，以上就是“Flink Connectors怎么连接MySql”的内容了，经过本文的学习后，相信大家对Flink Connectors怎么连接MySql这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！