Flink的sink实战之四:自定义

Flink的sink实战之四:自定义

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

本篇概览

Flink官方提供的sink服务可能满足不了我们的需要,此时可以开发自定义的sink,文本就来一起实战;

全系列链接

  1. 《Flink的sink实战之一:初探》
  2. 《Flink的sink实战之二:kafka》
  3. 《Flink的sink实战之三:cassandra3》
  4. 《Flink的sink实战之四:自定义》

继承关系

  1. 在正式编码前,要先弄清楚对sink能力是如何实现的,前面我们实战过的print、kafka、cassandra等sink操作,核心类的继承关系如下图所示:
    在这里插入图片描述
  2. 可见实现sink能力的关键,是实现RichFunction和SinkFunction接口,前者用于资源控制(如open、close等操作),后者负责sink的具体操作,来看看最简单的PrintSinkFunction类是如何实现SinkFunction接口的invoke方法:
@Override
public void invoke(IN record) {
  writer.write(record);
}
  1. 现在对sink的基本逻辑已经清楚了,可以开始编码实战了;

内容和版本

本次实战很简单:自定义sink,用于将数据写入MySQL,涉及的版本信息如下:

  1. jdk:1.8.0_191
  2. flink:1.9.2
  3. maven:3.6.0
  4. flink所在操作系统:CentOS Linux release 7.7.1908
  5. MySQL:5.7.29
  6. IDEA:2018.3.5 (Ultimate Edition)

源码下载

如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blog_demos):

名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址,https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址,ssh协议

这个git项目中有多个文件夹,本章的应用在flinksinkdemo文件夹下,如下图红框所示:
在这里插入图片描述

数据库准备

请您将MySQL准备好,并执行以下sql,用于创建数据库flinkdemo和表student:

create database if not exists flinkdemo;
USE flinkdemo;
DROP TABLE IF EXISTS `student`;
CREATE TABLE `student` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `name` 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;

编码

  1. 使用《Flink的sink实战之二:kafka》中创建的flinksinkdemo工程;
  2. 在pom.xml中增加mysql的依赖:
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>8.0.11</version>
</dependency>
  1. 创建和数据库的student表对应的实体类Student.java:
package com.bolingcavalry.customize;

public class Student {
    private int id;
    private String name;
    private int 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 int getAge() {
        return age;
    }

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

    public Student(String name, int age) {
        this.name = name;
        this.age = age;
    }
}
  1. 创建自定义sink类MySQLSinkFunction.java,这是本文的核心,有关数据库的连接、断开、写入数据都集中在此:
package com.bolingcavalry.customize;

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;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class MySQLSinkFunction extends RichSinkFunction<Student> {

    PreparedStatement preparedStatement;

    private Connection connection;

    private ReentrantLock reentrantLock = new ReentrantLock();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        //准备数据库相关实例
        buildPreparedStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();

        try{
            if(null!=preparedStatement) {
                preparedStatement.close();
                preparedStatement = null;
            }
        } catch(Exception e) {
            e.printStackTrace();
        }

        try{
            if(null!=connection) {
                connection.close();
                connection = null;
            }
        } catch(Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void invoke(Student value, Context context) throws Exception {
        preparedStatement.setString(1, value.getName());
        preparedStatement.setInt(2, value.getAge());
        preparedStatement.executeUpdate();
    }

    /**
     * 准备好connection和preparedStatement
     * 获取mysql连接实例,考虑多线程同步,
     * 不用synchronize是因为获取数据库连接是远程操作,耗时不确定
     * @return
     */
    private void buildPreparedStatement() {
        if(null==connection) {
            boolean hasLock = false;
            try {
                hasLock = reentrantLock.tryLock(10, TimeUnit.SECONDS);

                if(hasLock) {
                    Class.forName("com.mysql.cj.jdbc.Driver");
                    connection = DriverManager.getConnection("jdbc:mysql://192.168.50.43:3306/flinkdemo?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=UTC", "root", "123456");
                }

                if(null!=connection) {
                    preparedStatement = connection.prepareStatement("insert into student (name, age) values (?, ?)");
                }
            } catch (Exception e) {
                //生产环境慎用
                e.printStackTrace();
            } finally {
                if(hasLock) {
                    reentrantLock.unlock();
                }
            }
        }
    }
}
  1. 上述代码很简单,只需要注意在创建连接的时候用到了锁来控制多线程同步,以及高版本mysql驱动对应的driver和uri的写法与以前5.x版本的区别;
  2. 创建任务类StudentSink.java,用来创建一个flink任务,里面通过ArrayList创建了一个数据集,然后直接addSink,为了看清DAG,调用disableChaining方法取消了operator chain:
package com.bolingcavalry.customize;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.ArrayList;
import java.util.List;

public class StudentSink {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //并行度为1
        env.setParallelism(1);

        List<Student> list = new ArrayList<>();
        list.add(new Student("aaa", 11));
        list.add(new Student("bbb", 12));
        list.add(new Student("ccc", 13));
        list.add(new Student("ddd", 14));
        list.add(new Student("eee", 15));
        list.add(new Student("fff", 16));

        env.fromCollection(list)
            .addSink(new MySQLSinkFunction())
            .disableChaining();

        env.execute("sink demo : customize mysql obj");
    }
}
  1. 在flink web页面提交任务,并设置任务类:
    在这里插入图片描述
  2. 任务完成后,DAG图显示任务和记录数都符合预期:
    在这里插入图片描述
  3. 去检查数据库,发现数据已写入:

在这里插入图片描述

至此,自定义sink的实战已经完成,希望本文能给您一些参考;

欢迎关注公众号:程序员欣宸

微信搜索「程序员欣宸」,我是欣宸,期待与您一同畅游Java世界…
https://github.com/zq2599/blog_demos

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/2630.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • 什么叫做公网IP_是不是公网ip

    什么叫做公网IP_是不是公网ip由于公网ip资源匮乏,NAT(地址转换)技术被广泛应用。其好处就是让更多的电脑能够上网,缺点在于你自己搭建了一台服务器。想实现远程访问,如果服务器的地址如果是经过NAT转换后的地址,外网是无法访问到的。这就引出了公网ip和私网ip的概念,可以通过开头的数字来判断ip地址的类型,下面就给大家普及一下。由于私网ip地址仅限于局域网内使用,并且是可以重复的,所以IANA当初划分了一些网段,专供局域网内使用。具体网段如下:10.x.x.x192.168.x.x172.16.x.x-172.3

    2022年10月21日
  • eclipse代码自动补全[通俗易懂]

    eclipse代码自动补全[通俗易懂]1、点击菜单栏,打开Eclipse->Window->Perferences2、找到Java下的 Editor下的 ContentAssist,点击它3、找到第二个“AutoactivationtriggersforJava:”选项,在其后的文本框中会看到一个“.”存在。这表示:只有输入“.”之后才会有代码提示和自动补全,把该文本框中的“.”换成“abcdefghijklmnopqrstuvwxyz.”即可。…

  • python 写的一款百度关键字url多线程采集

    python 写的一款百度关键字url多线程采集

  • A Painless Q-learning Tutorial (一个 Q-learning 算法的简明教程)

    本文是对 http://mnemstudio.org/path-finding-q-learning-tutorial.htm 的翻译,共分两部分,第一部分为中文翻译,第二部分为英文原文。翻译时为方便读者理解,有些地方采用了意译的方式,此外,原文中有几处笔误,在翻译时已进行了更正。这篇教程通俗易懂,是学习理解Q-learning算法工作原理的绝佳入门材料。

  • PrintWriter and BufferedWriter区别和使用「建议收藏」

    PrintWriter and BufferedWriter区别和使用「建议收藏」区别:BufferedWriter:将文本写入字符输出流,缓冲各个字符从而提供单个字符,数组和字符串的高效写入。通过write()方法可以将获取到的字符输出,然后通过newLine()进行换行操作。并且BufferedWriter只能对字符流进行操作。PrintWriter:相对于BufferedWriter的好处在于,如果PrintWriter开启了自动刷新,那么当PrintWri…

  • eclipse环境下spring整合mybatis详细教程[通俗易懂]

    eclipse环境下spring整合mybatis详细教程[通俗易懂]系列目录第一篇:3分钟快速了解Mybatis的基础配置第二篇:带你3分钟了解Mybatis映射文件(sql,resultMap等映射)第三篇:三分钟带你了解mybatis关联映射(案例分析一对一,多对多)原创不易,如若喜欢,就点一点赞,关注一下吧!文章目录系列目录一、整合环境搭建-jar包准备1.spring所需要使用的jar包有(8+2):2.mybatis所需要使用的jar包有3.spring整合mybatis的中间jar二、整合环境搭建-创建项目1.eclipse环境创建2.jar添

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号