Flink的sink實戰之四：自定義

時間 2021-04-04

標籤 java mysql git 程序員 github web sql 數據庫 apache api 欄目 Java 简体版

原文原文鏈接

歡迎訪問個人GitHub

https://github.com/zq2599/blog_demosjava

內容：全部原創文章分類彙總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；mysql

本篇概覽

Flink官方提供的sink服務可能知足不了咱們的須要，此時能夠開發自定義的sink，文本就來一塊兒實戰；git

全系列連接

繼承關係

在正式編碼前，要先弄清楚對sink能力是如何實現的，前面咱們實戰過的print、kafka、cassandra等sink操做，核心類的繼承關係以下圖所示：

可見實現sink能力的關鍵，是實現<font color="blue">RichFunction和SinkFunction</font>接口，前者用於資源控制（如open、close等操做），後者負責sink的具體操做，來看看最簡單的PrintSinkFunction類是如何實現SinkFunction接口的invoke方法：

@Override
public void invoke(IN record) {
  writer.write(record);
}

如今對sink的基本邏輯已經清楚了，能夠開始編碼實戰了；

內容和版本

本次實戰很簡單：自定義sink，用於將數據寫入MySQL，涉及的版本信息以下：程序員

jdk：1.8.0_191
flink：1.9.2
maven：3.6.0
flink所在操做系統：CentOS Linux release 7.7.1908
MySQL：5.7.29
IDEA：2018.3.5 (Ultimate Edition)

源碼下載

若是您不想寫代碼，整個系列的源碼可在GitHub下載到，地址和連接信息以下表所示(https://github.com/zq2599/blo...：github

名稱	連接	備註
項目主頁	https://github.com/zq2599/blo...	該項目在GitHub上的主頁
git倉庫地址(https)	https://github.com/zq2599/blo...	該項目源碼的倉庫地址，https協議
git倉庫地址(ssh)	git@github.com:zq2599/blog_demos.git	該項目源碼的倉庫地址，ssh協議

這個git項目中有多個文件夾，本章的應用在<font color="blue">flinksinkdemo</font>文件夾下，以下圖紅框所示：
web

數據庫準備

請您將MySQL準備好，並執行如下sql，用於建立數據庫flinkdemo和表student：sql

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;

編碼

使用《Flink的sink實戰之二：kafka》中建立的flinksinkdemo工程；
在pom.xml中增長mysql的依賴：

<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>8.0.11</version>
</dependency>

建立和數據庫的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;
    }
}

建立自定義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();
                }
            }
        }
    }
}

上述代碼很簡單，只須要注意在建立鏈接的時候用到了鎖來控制多線程同步，以及高版本mysql驅動對應的driver和uri的寫法與之前5.x版本的區別；
建立任務類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");
    }
}