Flink的DataSource三部曲之三:自定義

時間  2020-12-03
標籤 java mysql git 程序員 github web sql docker shell 數據庫 欄目 Java 简体版
原文   原文鏈接

本文是《Flink的DataSource三部曲》的終篇，前面都是在學習Flink已有的數據源功能，但若是這些不能知足須要，就要自定義數據源（例如從數據庫獲取數據），也就是今天實戰的內容，以下圖紅框所示：
在這裏插入圖片描述### Flink的DataSource三部曲文章連接
java

  1. 《Flink的DataSource三部曲之一：直接API》
  2. 《Flink的DataSource三部曲之二:內置connector》
  3. 《Flink的DataSource三部曲之三:自定義》

環境和版本

本次實戰的環境和版本以下：mysql

  1. JDK：1.8.0_211
  2. Flink：1.9.2
  3. Maven：3.6.0
  4. 操做系統：macOS Catalina 10.15.3 （MacBook Pro 13-inch, 2018）
  5. IDEA：2018.3.5 (Ultimate Edition)

在服務器上搭建Flink服務

  1. 前面兩章的程序都是在IDEA上運行的，本章須要經過Flink的web ui觀察運行結果，所以要單獨部署Flink服務，我這裏是在CentOS環境經過docker-compose部署的，如下是docker-compose.yml的內容，用於參考：
version: "2.1"
services:
  jobmanager:
    image: flink:1.9.2-scala_2.12
    expose:
      - "6123"
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
  taskmanager1:
    image: flink:1.9.2-scala_2.12
    expose:
      - "6121"
      - "6122"
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
  taskmanager2:
    image: flink:1.9.2-scala_2.12
    expose:
      - "6121"
      - "6122"
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
  1. 下圖是個人Flink狀況，有兩個Task Maganer，共八個Slot所有可用：
    在這裏插入圖片描述

源碼下載

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

名稱 連接 備註
項目主頁 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項目中有多個文件夾，本章的應用在flinkdatasourcedemo文件夾下，以下圖紅框所示：
在這裏插入圖片描述
準備完畢，開始開發；

程序員

實現SourceFunctionDemo接口的DataSource

  1. 從最簡單的開始，開發一個不可並行的數據源並驗證；
  2. 實現SourceFunction接口，在工程flinkdatasourcedemo中增長SourceFunctionDemo.java：
package com.bolingcavalry.customize;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;

public class SourceFunctionDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //並行度爲2
        env.setParallelism(2);

        DataStream<Tuple2<Integer,Integer>> dataStream = env.addSource(new SourceFunction<Tuple2<Integer, Integer>>() {

            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
                int i = 0;
                while (isRunning) {
                    ctx.collect(new Tuple2<>(i++ % 5, 1));
                    Thread.sleep(1000);
                    if(i>9){
                        break;
                    }
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        dataStream
                .keyBy(0)
                .timeWindow(Time.seconds(2))
                .sum(1)
                .print();

        env.execute("Customize DataSource demo : SourceFunction");
    }
}
  1. 從上述代碼可見，給addSource方法傳入一個匿名類實例，該匿名類實現了SourceFunction接口；
  2. 實現SourceFunction接口只需實現run和cancel方法；
  3. run方法產生數據，這裏爲了簡答操做，每隔一秒產生一個Tuple2實例，因爲接下來的算子中有keyBy操做，所以Tuple2的第一個字段始終保持着5的餘數，這樣能夠多幾個key，以便分散到不一樣的slot中；
  4. 爲了覈對數據是否準確，這裏並無無限發送數據，而是僅發送了10個Tuple2實例；
  5. cancel是job被取消時執行的方法；
  6. 總體並行度顯式設置爲2；
  7. 編碼完成後，執行mvn clean package -U -DskipTests構建，在target目錄獲得文件flinkdatasourcedemo-1.0-SNAPSHOT.jar；
  8. 在Flink的web UI上傳flinkdatasourcedemo-1.0-SNAPSHOT.jar，並指定執行類，以下圖紅框所示：
    在這裏插入圖片描述
  9. 任務執行完成後，在Completed Jobs頁面能夠看到，DataSource的並行度是1（紅框），對應的SubTask一共發送了10條記錄（藍框），這和咱們的代碼是一致的；
    在這裏插入圖片描述
  10. 再來看消費的子任務，以下圖，紅框顯示並行度是2，這和前面代碼中的設置是一致的，藍框顯示兩個子任務一共收到10條數據記錄，和上游發出的數量一致：
    在這裏插入圖片描述
  11. 接下來嘗試多並行度的DataSource；

實現ParallelSourceFunction接口的DataSource

  1. 若是自定義DataSource中有複雜的或者耗時的操做，那麼增長DataSource的並行度，讓多個SubTask同時進行這些操做，能夠有效提高總體吞吐量（前提是硬件資源充裕）；
  2. 接下來實戰能夠並行執行的DataSource，原理是DataSoure實現ParallelSourceFunction接口，代碼以下，可見和SourceFunctionDemo幾乎同樣，只是addSource方發入參不一樣，該入參依然是匿名類，不過實現的的接口變成了ParallelSourceFunction：
package com.bolingcavalry.customize;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;

public class ParrelSourceFunctionDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //並行度爲2
        env.setParallelism(2);

        DataStream<Tuple2<Integer,Integer>> dataStream = env.addSource(new ParallelSourceFunction<Tuple2<Integer, Integer>>() {

            private volatile boolean isRunning = true;

            @Override
            public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
                int i = 0;
                while (isRunning) {
                    ctx.collect(new Tuple2<>(i++ % 5, 1));
                    Thread.sleep(1000);
                    if(i>9){
                        break;
                    }
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        dataStream
                .keyBy(0)
                .timeWindow(Time.seconds(2))
                .sum(1)
                .print();

        env.execute("Customize DataSource demo : ParallelSourceFunction");
    }
}
  1. 編碼完成後，執行mvn clean package -U -DskipTests構建，在target目錄獲得文件flinkdatasourcedemo-1.0-SNAPSHOT.jar；
  2. 在Flink的web UI上傳flinkdatasourcedemo-1.0-SNAPSHOT.jar，並指定執行類，以下圖紅框所示：
    在這裏插入圖片描述
  3. 任務執行完成後，在Completed Jobs頁面能夠看到，現在DataSource的並行度是2（紅框），對應的SubTask一共發送了20條記錄（藍框），這和咱們的代碼是一致的，綠框顯示兩個SubTask的Task Manager是同一個：
    在這裏插入圖片描述
  4. 爲何DataSource一共發送了20條記錄？由於每一個SubTask中都有一份ParallelSourceFunction匿名類的實例，對應的run方法分別被執行，所以每一個SubTask都發送了10條；
  5. 再來看消費數據的子任務，以下圖，紅框顯示並行度與代碼中設置的數量是一致的，藍框顯示兩個SubTask一共消費了20條記錄，和數據源發出的記錄數一致，另外綠框顯示兩個SubTask的Task Manager是同一個，並且和DataSource的TaskManager是同一個，所以整個job都是在同一個TaskManager進行的，沒有跨機器帶來的額外代價：
    在這裏插入圖片描述
  6. 接下來要實踐的內容，和另外一個重要的抽象類有關；

繼承抽象類RichSourceFunction的DataSource

  1. 對RichSourceFunction的理解是從繼承關係開始的，以下圖，SourceFunction和RichFunction的特性最終都體如今RichSourceFunction上，SourceFunction的特性是數據的生成(run方法)，RichFunction的特性是對資源的鏈接和釋放(open和close方法)：
    在這裏插入圖片描述
  2. 接下來開始實戰，目標是從MySQL獲取數據做爲DataSource，而後消費這些數據；
  3. 請提早準備好可用的MySql數據庫，而後執行如下SQL，建立庫、表、記錄：
DROP DATABASE IF EXISTS flinkdemo;
CREATE DATABASE IF NOT EXISTS flinkdemo;
USE flinkdemo;

SELECT 'CREATING DATABASE STRUCTURE' as 'INFO';

DROP TABLE IF EXISTS `student`;
CREATE TABLE `student` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(25) COLLATE utf8_bin DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;

INSERT INTO `student` VALUES ('1', 'student01'), ('2', 'student02'), ('3', 'student03'), ('4', 'student04'), ('5', 'student05'), ('6', 'student06');
COMMIT;
  1. 在pom.xml中增長mysql依賴：
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.34</version>
</dependency>
  1. 新增MySQLDataSource.java，內容以下：
package com.bolingcavalry.customize;

import com.bolingcavalry.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class MySQLDataSource extends RichSourceFunction<Student> {

    private Connection connection = null;

    private PreparedStatement preparedStatement = null;

    private volatile boolean isRunning = true;

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

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

        if(null==preparedStatement) {
            preparedStatement = connection.prepareStatement("select id, name from student");
        }
    }

    /** * 釋放資源 * @throws Exception */
    @Override
    public void close() throws Exception {
        super.close();

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

        if(null==connection) {
            connection.close();
        }
    }

    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        ResultSet resultSet = preparedStatement.executeQuery();
        while (resultSet.next() && isRunning) {
            Student student = new Student();
            student.setId(resultSet.getInt("id"));
            student.setName(resultSet.getString("name"));
            ctx.collect(student);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
  1. 上面的代碼中，MySQLDataSource繼承了RichSourceFunction，做爲一個DataSource，能夠做爲addSource方法的入參；
  2. open和close方法都會被數據源的SubTask調用，open負責建立數據庫鏈接對象，close負責釋放資源；
  3. open方法中直接寫死了數據庫相關的配置(不可取)；
  4. run方法在open以後被調用，做用和以前的DataSource例子同樣，負責生產數據，這裏是用前面準備好的preparedStatement對象直接去數據庫取數據；
  5. 接下來寫個Demo類使用MySQLDataSource：
package com.bolingcavalry.customize;

import com.bolingcavalry.Student;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RichSourceFunctionDemo {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //並行度爲2
        env.setParallelism(2);

        DataStream<Student> dataStream = env.addSource(new MySQLDataSource());
        dataStream.print();

        env.execute("Customize DataSource demo : RichSourceFunction");
    }
}
  1. 從上述代碼可見，MySQLDataSource實例傳入addSource方法便可建立數據集；
  2. 像以前那樣，編譯構建、提交到Flink、指定任務類，便可開始執行此任務；
  3. 執行結果以下圖，DataSource的並行度是1，一共發送六條記錄，即student表的全部記錄：
    在這裏插入圖片描述
  4. 處理數據的SubTask一共兩個，各處理三條消息：
    在這裏插入圖片描述
  5. 因爲代碼中對數據集執行了print()，所以在TaskManager控制檯看到數據輸出以下圖紅框所示：
    在這裏插入圖片描述

關於RichParallelSourceFunction

  1. 實戰到了這裏，還剩RichParallelSourceFunction這個抽象類咱們尚未嘗試過，但我以爲這個類能夠不用在文中多說了，我們把RichlSourceFunction和RichParallelSourceFunction的類圖放在一塊兒看看：
    在這裏插入圖片描述
  2. 從上圖可見，在RichFunction繼承關係上，二者一致，在SourceFunction的繼承關係上，RichlSourceFunction和RichParallelSourceFunction略有不一樣，RichParallelSourceFunction走的是ParallelSourceFunction這條線，而SourceFunction和ParallelSourceFunction的區別，前面已經講過了，所以，結果不言而喻：繼承RichParallelSourceFunction的DataSource的並行度是能夠大於1的；
  3. 讀者您若是有興趣，能夠將前面的MySQLDataSource改爲繼承RichParallelSourceFunction再試試，DataSource的並行度會超過1，可是毫不是隻有這一點變化，DAG圖顯示Flink還會作一些Operator Chain處理，但這不是本章要關注的內容，只能說結果是正確的（兩個DataSource的SubTask，一共發送12條記錄），建議您試試；

至此，《Flink的DataSource三部曲》系列就所有完成了，好的開始是成功的一半，在拿到數據後，後面還有不少知識點要學習和掌握，接下來的文章會繼續深刻Flink的奇妙之旅；github

歡迎關注個人公衆號：程序員欣宸

在這裏插入圖片描述

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程