Flink RichSourceFunction应用，读关系型数据(mysql)数据写入关系型数据库(mysql)

1. 写在前面

Flink被誉为第四代大数据计算引擎组件，即可以用作基于离线分布式计算，也可以应用于实时计算。Flink的核心是转化为流进行计算。Flink三个核心：Source,Transformation,Sink。其中Source即为Flink计算的数据源，Transformation即为进行分布式流式计算的算子，也是计算的核心，Sink即为计算后的数据输出端。Flink Source原生支持包括Kafka,ES,RabbitMQ等一些通用的消息队列组件或基于文本的高性能非关系型数据库。而Flink Sink写原生也只支持类似Redis，Kafka,ES,RabbitMQ等一些通用的消息队列组件或基于文本的高性能非关系型数据库。而对于写入关系型数据库或Flink不支持的组件中，需要借助RichSourceFunction去实现，但这部分性能是比原生的差些，虽然Flink不建议这么做，但在大数据处理过程中，由于业务或技术架构的复杂性，有些特定的场景还是需要这样做，本篇博客就是介绍如何通过Flink RichSourceFunction来写关系型数据库，这里以写mysql为例。

2. 引入依赖的jar包

flink基础包
flink-jdbc包
mysql-jdbc包

3. 继承RichSourceFunction包将jdbc封装读mysql

package com.run;

import java.sql.DriverManager;
import java.sql.ResultSet;

import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class Flink2JdbcReader extends
        RichSourceFunction<Tuple10<String, String, String, String, String, String, String, String, String, String>> {
    private static final long serialVersionUID = 3334654984018091675L;

    private Connection connect = null;
    private PreparedStatement ps = null;

    /*
     * (non-Javadoc)
     *
     * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
     * apache.flink.configuration.Configuration) to use open database connect
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.21.11:3306", "root", "flink");
        ps = (PreparedStatement) connect
                .prepareStatement("select col1,col2,col3,col4,col5,col6,col7,col8,col9,col10 from flink.test_tb");
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.apache.flink.streaming.api.functions.source.SourceFunction#run(org.
     * apache.flink.streaming.api.functions.source.SourceFunction.SourceContext)
     * to use excuted sql and return result
     */
    @Override
    public void run(
            SourceContext<Tuple10<String, String, String, String, String, String, String, String, String, String>> collect)
            throws Exception {
        ResultSet resultSet = ps.executeQuery();
        while (resultSet.next()) {
            Tuple10<String, String, String, String, String, String, String, String, String, String> tuple = new Tuple10<String, String, String, String, String, String, String, String, String, String>();
            tuple.setFields(resultSet.getString(1), resultSet.getString(2), resultSet.getString(3),
                    resultSet.getString(4), resultSet.getString(5), resultSet.getString(6), resultSet.getString(7),
                    resultSet.getString(8), resultSet.getString(9), resultSet.getString(10));
            collect.collect(tuple);
        }

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.apache.flink.streaming.api.functions.source.SourceFunction#cancel()
     * colse database connect
     */
    @Override
    public void cancel() {
        try {
            super.close();
            if (connect != null) {
                connect.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}

4. 继承RichSourceFunction包将jdbc封装写mysql

package com.run;

import java.sql.DriverManager;

import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;

public class Flink2JdbcWriter extends
        RichSinkFunction<Tuple10<String, String, String, String, String, String, String, String, String, String>> {
    private static final long serialVersionUID = -8930276689109741501L;

    private Connection connect = null;
    private PreparedStatement ps = null;

    /*
     * (non-Javadoc)
     *
     * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
     * apache.flink.configuration.Configuration) get database connect
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.21.11:3306", "root", "flink");
        ps = (PreparedStatement) connect.prepareStatement("insert into flink.test_tb1 values (?,?,?,?,?,?,?,?,?,?)");
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.apache.flink.streaming.api.functions.sink.SinkFunction#invoke(java.
     * lang.Object,
     * org.apache.flink.streaming.api.functions.sink.SinkFunction.Context) read
     * data from flink DataSet to database
     */
    @Override
    public void invoke(Tuple10<String, String, String, String, String, String, String, String, String, String> value,
            Context context) throws Exception {
        ps.setString(1, value.f0);
        ps.setString(2, value.f1);
        ps.setString(3, value.f2);
        ps.setString(4, value.f3);
        ps.setString(5, value.f4);
        ps.setString(6, value.f5);
        ps.setString(7, value.f6);
        ps.setString(8, value.f7);
        ps.setString(9, value.f8);
        ps.setString(10, value.f9);
        ps.executeUpdate();
    }

    /*
     * (non-Javadoc)
     *
     * @see org.apache.flink.api.common.functions.AbstractRichFunction#close()
     * close database connect
     */
    @Override
    public void close() throws Exception {
        try {
            super.close();
            if (connect != null) {
                connect.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4. 代码解释

对于Flink2JdbcReader的读
里面有三个方法open，run，cancel，其中open方法是建立与关系型数据库的链接，这里其实就是普通的jdbc链接及mysql的地址，端口，库等信息。run方法是读取mysql数据转化为Flink独有的Tuple集合类型，可以根据代码看出其中的规律和Tuple8，Tuple9，Tuple10代表什么含义。cancel就很简单了关闭数据库连接

对于Flink2JdbcWriter的写
里面有三个方法open，invoke，close，其中open方法是建立与关系型数据库的链接，这里其实就是普通的jdbc链接及mysql的地址，端口，库等信息。invoke方法是将flink的数据类型插入到mysql，这里的写法与在web程序中写jdbc插入数据不太一样，因为flink独有的Tuple，可以根据代码看出其中的规律和Tuple8，Tuple9，Tuple10代表什么含义。close关闭数据库连接

5. 测试:读mysql数据并继续写入mysql

package com.run;

import java.util.Date;

import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkReadDbWriterDb {
    public static void main(String[] args) throws Exception {。
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple10<String, String, String, String, String, String, String, String, String, String>> dataStream = env
                .addSource(new Flink2JdbcReader());

        // tranfomat

        dataStream.addSink(new Flink2JdbcWriter());
        env.execute("Flink cost DB data to write Database");

    }
}

6. 总结

从测试代码中可以很清晰的看出Flink的逻辑:Source->Transformation->Sink,可以在addSource到addSink之间加入我们的业务逻辑算子。同时这里必须注意env.execute("Flink cost DB data to write Database");这个必须有而且必须要放到结尾，否则整个代码是不会执行的，至于为什么在后续的博客会讲

原文地址：https://www.cnblogs.com/jiashengmei/p/10567343.html

时间： 2024-08-29 22:00:40

Flink RichSourceFunction应用，读关系型数据(mysql)数据写入关系型数据库(mysql)的相关文章

MYSQL添加新用户 MYSQL为用户创建数据库 MYSQL为新用户分配权限

1.新建用户 //登录MYSQL@>mysql -u root -p@>密码//创建用户mysql> insert into mysql.user(Host,User,Password) values('localhost','jeecn',password('jeecn'));//刷新系统权限表mysql>flush privileges;这样就创建了一个名为:jeecn  密码为:jeecn  的用户. //退出后登录一下mysql>exit;@>mysql -u

迁移 RDS for MySQL 数据到本地 MySQL

背景  公司的业务系统采用阿里云的RDS云数据库,最近有一个需求:迁移 RDS for MySQL 数据到本地 MySQL.阿里云数据库 MySQL 版支持通过物理备份文件和逻辑备份文件两种途径将云上数据迁移到本地数据库.我公司采用mysql5.6版,备分策略是自动全量物理备份.本地系统是Centos 6.8 X64,此教程提供两种方法,任选其一即可. 备注:因为软件限制,目前数据恢复只支持在 Linux 系统下进行.如果您要恢复数据到 Windows 系统,可以先将数据恢复到 Linux 系统

mysql数据库命令大全---完全备份和恢复

数据完全备份[[email protected] ~]# mysqldump -uroot -pabc123 --databases school > /opt/school.sql 删除库mysql> drop database school;查看数据库mysql> show databases; +--------------------+ | Database | +--------------------+ | information_schema | | mysql | | p

mysql数据语句

SELECT fl.name,fvl.value,COUNT((fvl.value)) n  FROM ps_feature_lang  fl LEFT JOIN ps_feature_product fp  ON (fl.id_feature=fp.id_feature) LEFT JOIN ps_feature_value_lang fvl ON (fvl.id_feature_value=fp.id_feature_value)  WHERE fp.id_product in (selec

ODP方式，大批量数据写入ORACLE数据库

项目中在同步数据的时候,需要把获得的数据DataTable,写入oracle数据库 因为System.Data.OracleClient写入方式写入大批量数据特别慢,改用Oracle.DataAccess写入方式(上代码): ODP工具类: 需要引入命名空间: using Oracle.DataAccess;using Oracle.DataAccess.Client;using Oracle.DataAccess.Types; 1 ODP_Inserter 2 { 3 /// <summary

MySQL常用命令（数据库，表相关的命令）

数据库相关命令 显示数据库列表 mysql> SHOW  DATABASES; 创建数据库 mysql> CREATE  DATABASE  库名; 如下,创建一个名为crashcourse的数据库 mysql> CREATE  DATABASE  crashcourse; 删除数据库 mysql> DROP  DATABASE  库名; 如下,删除一个名为crashcourse的数据库 mysql> DROP  DATABASE  crashcourse; 使用数据库 my

Python3爬虫（九） 数据存储之关系型数据库MySQL

Infi-chu: http://www.cnblogs.com/Infi-chu/ 关系型数据库关系型数据库是基于关系模型的数据库,而关系模型是通过二维表来保存的,所以关系型数据库的存储方式就是行列组成的表.每一列是一个字段,每一行是一条记录.表可以看成是摸个实体的集合,而实体之间存在关系,比如主键和外键常见的关系型数据库有:Oracle.MySQL.SQLite.SQL Server.DB2等 MySQL在Python2中,MySQL的链接库是MySQLdb, 而在Python3中,官方不支

Hadoop 中利用 mapreduce 读写 mysql 数据

Hadoop 中利用 mapreduce 读写 mysql 数据 有时候我们在项目中会遇到输入结果集很大,但是输出结果很小,比如一些 pv.uv 数据,然后为了实时查询的需求,或者一些 OLAP 的需求,我们需要 mapreduce 与 mysql 进行数据的交互,而这些特性正是 hbase 或者 hive 目前亟待改进的地方. 好了言归正传,简单的说说背景.原理以及需要注意的地方: 1.为了方便 MapReduce 直接访问关系型数据库(Mysql,Oracle),Hadoop提供了DBInp

MySQL数据库面试题

最近在换工作,虽然面试的不是dba,但是一些数据库的知识还是会有问到,然后百度了一下,结合我搜到的和被问到的总结一下. 1.mysql支持事务吗? 在缺省的模式下,是自动提交的,所有的数据库更新操作都会即时提交,但是当将表类型使用innodb或bdb就可进行事务处理: set  autocommit=0: start transaction; select @a:=count(id) from table1 where name='aa'; update table2 set [email pr