• 150606

    文章

  • 1200

    评论

  • 13

    友链

  • 最近新加了换肤功能,大家多来逛逛吧~~~~
  • 喜欢这个网站的朋友可以加一下QQ群,我们一起交流技术。

Flink 系例 之 Connectors 连接 MySql


通过使用Flink DataStream Connectors 数据流连接器连接到Mysql数据源,并基于JDBC提供数据流输入与输出操作;

示例环境

java.version: 1.8.x
flink.version: 1.11.1
mysql:5.7.x

示例数据源 (项目码云下载)

Flink 系例 之 搭建开发环境与数据

示例模块 (pom.xml)

Flink 系例 之 DataStream Connectors 与 示例模块

数据流输入

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";

}

数据展示


695856371Web网页设计师②群 | 喜欢本站的朋友可以收藏本站,或者加入我们大家一起来交流技术!

0条评论

Loading...


发表评论

电子邮件地址不会被公开。 必填项已用*标注

自定义皮肤 主体内容背景
打开支付宝扫码付款购买视频教程
遇到问题联系客服QQ:419400980
注册梁钟霖个人博客