Flink Connectors怎么連接MySql

這篇文章主要講解了“Flink Connectors怎么連接MySql”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Flink Connectors怎么連接MySql”吧！

通過使用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<TUser> 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<TUser> 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這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！