Flink 从 0 到 1 学习 —— 如何自定义 Data Sink ？

程序员文章站 2022-04-14 23:30:05

前言前篇文章 "《从0到1学习Flink》—— Data Sink 介绍" 介绍了 Flink Data Sink，也介绍了 Flink 自带的 Sink，那么如何自定义自己的 Sink 呢？这篇文章将写一个 demo 教大家将从 Kafka Source 的数据 Sink 到 MySQL 中去。 ......

前言

前篇文章《从0到1学习flink》—— data sink 介绍介绍了 flink data sink，也介绍了 flink 自带的 sink，那么如何自定义自己的 sink 呢？这篇文章将写一个 demo 教大家将从 kafka source 的数据 sink 到 mysql 中去。

准备工作

我们先来看下 flink 从 kafka topic 中获取数据的 demo，首先你需要安装好了 flink 和 kafka 。

运行启动 flink、zookepeer、kafka，

Flink 从 0 到 1 学习 —— 如何自定义 Data Sink ？

好了，都启动了！

数据库建表

drop table if exists `student`;
create table `student` (
  `id` int(11) unsigned not null auto_increment,
  `name` varchar(25) collate utf8_bin default null,
  `password` varchar(25) collate utf8_bin default null,
  `age` int(10) default null,
  primary key (`id`)
) engine=innodb auto_increment=5 default charset=utf8 collate=utf8_bin;

实体类

student.java

package com.zhisheng.flink.model;

/**
 * desc:
 * weixin: zhisheng_tian
 * blog: http://www.54tianzhisheng.cn/
 */
public class student {
    public int id;
    public string name;
    public string password;
    public int age;

    public student() {
    }

    public student(int id, string name, string password, int age) {
        this.id = id;
        this.name = name;
        this.password = password;
        this.age = age;
    }

    @override
    public string tostring() {
        return "student{" +
                "id=" + id +
                ", name='" + name + '\'' +
                ", password='" + password + '\'' +
                ", age=" + age +
                '}';
    }

    public int getid() {
        return id;
    }

    public void setid(int id) {
        this.id = id;
    }

    public string getname() {
        return name;
    }

    public void setname(string name) {
        this.name = name;
    }

    public string getpassword() {
        return password;
    }

    public void setpassword(string password) {
        this.password = password;
    }

    public int getage() {
        return age;
    }

    public void setage(int age) {
        this.age = age;
    }
}

工具类

工具类往 kafka topic student 发送数据

import com.alibaba.fastjson.json;
import com.zhisheng.flink.model.metric;
import com.zhisheng.flink.model.student;
import org.apache.kafka.clients.producer.kafkaproducer;
import org.apache.kafka.clients.producer.producerrecord;

import java.util.hashmap;
import java.util.map;
import java.util.properties;

/**
 * 往kafka中写数据
 * 可以使用这个main函数进行测试一下
 * weixin: zhisheng_tian
 * blog: http://www.54tianzhisheng.cn/
 */
public class kafkautils2 {
    public static final string broker_list = "localhost:9092";
    public static final string topic = "student";  //kafka topic 需要和 flink 程序用同一个 topic

    public static void writetokafka() throws interruptedexception {
        properties props = new properties();
        props.put("bootstrap.servers", broker_list);
        props.put("key.serializer", "org.apache.kafka.common.serialization.stringserializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.stringserializer");
        kafkaproducer producer = new kafkaproducer<string, string>(props);

        for (int i = 1; i <= 100; i++) {
            student student = new student(i, "zhisheng" + i, "password" + i, 18 + i);
            producerrecord record = new producerrecord<string, string>(topic, null, null, json.tojsonstring(student));
            producer.send(record);
            system.out.println("发送数据: " + json.tojsonstring(student));
        }
        producer.flush();
    }

    public static void main(string[] args) throws interruptedexception {
        writetokafka();
    }
}

sinktomysql

该类就是 sink function，继承了 richsinkfunction ，然后重写了里面的方法。在 invoke 方法中将数据插入到 mysql 中。

package com.zhisheng.flink.sink;

import com.zhisheng.flink.model.student;
import org.apache.flink.configuration.configuration;
import org.apache.flink.streaming.api.functions.sink.richsinkfunction;

import java.sql.connection;
import java.sql.drivermanager;
import java.sql.preparedstatement;

/**
 * desc:
 * weixin: zhisheng_tian
 * blog: http://www.54tianzhisheng.cn/
 */
public class sinktomysql extends richsinkfunction<student> {
    preparedstatement ps;
    private connection connection;

    /**
     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接
     *
     * @param parameters
     * @throws exception
     */
    @override
    public void open(configuration parameters) throws exception {
        super.open(parameters);
        connection = getconnection();
        string sql = "insert into student(id, name, password, age) values(?, ?, ?, ?);";
        ps = this.connection.preparestatement(sql);
    }

    @override
    public void close() throws exception {
        super.close();
        //关闭连接和释放资源
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }

    /**
     * 每条数据的插入都要调用一次 invoke() 方法
     *
     * @param value
     * @param context
     * @throws exception
     */
    @override
    public void invoke(student value, context context) throws exception {
        //组装数据，执行插入操作
        ps.setint(1, value.getid());
        ps.setstring(2, value.getname());
        ps.setstring(3, value.getpassword());
        ps.setint(4, value.getage());
        ps.executeupdate();
    }

    private static connection getconnection() {
        connection con = null;
        try {
            class.forname("com.mysql.jdbc.driver");
            con = drivermanager.getconnection("jdbc:mysql://localhost:3306/test?useunicode=true&characterencoding=utf-8", "root", "root123456");
        } catch (exception e) {
            system.out.println("-----------mysql get connection has exception , msg = "+ e.getmessage());
        }
        return con;
    }
}

flink 程序

这里的 source 是从 kafka 读取数据的，然后 flink 从 kafka 读取到数据（json）后用阿里 fastjson 来解析成 student 对象，然后在 addsink 中使用我们创建的 sinktomysql，这样就可以把数据存储到 mysql 了。

package com.zhisheng.flink;

import com.alibaba.fastjson.json;
import com.zhisheng.flink.model.student;
import com.zhisheng.flink.sink.sinktomysql;
import org.apache.flink.api.common.serialization.simplestringschema;
import org.apache.flink.streaming.api.datastream.datastreamsource;
import org.apache.flink.streaming.api.datastream.singleoutputstreamoperator;
import org.apache.flink.streaming.api.environment.streamexecutionenvironment;
import org.apache.flink.streaming.api.functions.sink.printsinkfunction;
import org.apache.flink.streaming.connectors.kafka.flinkkafkaconsumer011;
import org.apache.flink.streaming.connectors.kafka.flinkkafkaproducer011;

import java.util.properties;

/**
 * desc:
 * weixin: zhisheng_tian
 * blog: http://www.54tianzhisheng.cn/
 */
public class main3 {
    public static void main(string[] args) throws exception {
        final streamexecutionenvironment env = streamexecutionenvironment.getexecutionenvironment();

        properties props = new properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.stringdeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.stringdeserializer");
        props.put("auto.offset.reset", "latest");

        singleoutputstreamoperator<student> student = env.addsource(new flinkkafkaconsumer011<>(
                "student",   //这个 kafka topic 需要和上面的工具类的 topic 一致
                new simplestringschema(),
                props)).setparallelism(1)
                .map(string -> json.parseobject(string, student.class)); //fastjson 解析字符串成 student 对象

        student.addsink(new sinktomysql()); //数据 sink 到 mysql

        env.execute("flink add sink");
    }
}

结果

运行 flink 程序，然后再运行 kafkautils2.java 工具类，这样就可以了。

如果数据插入成功了，那么我们查看下我们的数据库：

Flink 从 0 到 1 学习 —— 如何自定义 Data Sink ？

数据库中已经插入了 100 条我们从 kafka 发送的数据了。证明我们的 sinktomysql 起作用了。是不是很简单？

项目结构

怕大家不知道我的项目结构，这里发个截图看下：

Flink 从 0 到 1 学习 —— 如何自定义 Data Sink ？

最后

本文主要利用一个 demo，告诉大家如何自定义 sink function，将从 kafka 的数据 sink 到 mysql 中，如果你项目中有其他的数据来源，你也可以换成对应的 source，也有可能你的 sink 是到其他的地方或者其他不同的方式，那么依旧是这个套路：继承 richsinkfunction 抽象类，重写 invoke 方法。