《從0到1學習Flink》—— Data Sink 介紹

阿新 • • 發佈：2018-12-02

前言

再上一篇文章中《從0到1學習Flink》—— Data Source 介紹講解了 Flink Data Source ，那麼這裡就來講講 Flink Data Sink 吧。

首先 Sink 的意思是：

大概可以猜到了吧！Data sink 有點把資料儲存下來（落庫）的意思。

如上圖，Source 就是資料的來源，中間的 Compute 其實就是 Flink 乾的事情，可以做一系列的操作，操作完後就把計算後的資料結果 Sink 到某個地方。（可以是 MySQL、ElasticSearch、Kafka、Cassandra 等）。這裡我說下自己目前做告警這塊就是把 Compute 計算後的結果 Sink 直接告警出來了（傳送告警訊息到釘釘群、郵件、簡訊等），這個 sink 的意思也不一定非得說成要把資料儲存到某個地方去。其實官網用的 Connector 來形容要去的地方更合適，這個 Connector 可以有 MySQL、ElasticSearch、Kafka、Cassandra RabbitMQ 等。

Flink Data Sink

前面文章《從0到1學習Flink》—— Data Source 介紹介紹了 Flink Data Source 有哪些，這裡也看看 Flink Data Sink 支援的有哪些。

看下原始碼有哪些呢？

可以看到有 Kafka、ElasticSearch、Socket、RabbitMQ、JDBC、Cassandra POJO、File、Print 等 Sink 的方式。

SinkFunction

從上圖可以看到 SinkFunction 介面有 invoke 方法，它有一個 RichSinkFunction 抽象類。

上面的那些自帶的 Sink 可以看到都是繼承了 RichSinkFunction 抽象類，實現了其中的方法，那麼我們要是自己定義自己的 Sink 的話其實也是要按照這個套路來做的。

這裡就拿個較為簡單的 PrintSinkFunction 原始碼來講下：

@PublicEvolving
public class PrintSinkFunction<IN> extends RichSinkFunction<IN> {
    private static final long serialVersionUID = 1L;

    private static final boolean STD_OUT = false;
    private static final boolean STD_ERR = true;

    private boolean target;
    private transient PrintStream stream;
    private transient String prefix;

    /**
     * Instantiates a print sink function that prints to standard out.
     */
    public PrintSinkFunction() {}

    /**
     * Instantiates a print sink function that prints to standard out.
     *
     * @param stdErr True, if the format should print to standard error instead of standard out.
     */
    public PrintSinkFunction(boolean stdErr) {
        target = stdErr;
    }

    public void setTargetToStandardOut() {
        target = STD_OUT;
    }

    public void setTargetToStandardErr() {
        target = STD_ERR;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
        // get the target stream
        stream = target == STD_OUT ? System.out : System.err;

        // set the prefix if we have a >1 parallelism
        prefix = (context.getNumberOfParallelSubtasks() > 1) ?
                ((context.getIndexOfThisSubtask() + 1) + "> ") : null;
    }

    @Override
    public void invoke(IN record) {
        if (prefix != null) {
            stream.println(prefix + record.toString());
        }
        else {
            stream.println(record.toString());
        }
    }

    @Override
    public void close() {
        this.stream = null;
        this.prefix = null;
    }

    @Override
    public String toString() {
        return "Print to " + (target == STD_OUT ? "System.out" : "System.err");
    }
}

可以看到它就是實現了 RichSinkFunction 抽象類，然後實現了 invoke 方法，這裡 invoke 方法就是把記錄打印出來了就是，沒做其他的額外操作。

如何使用？

SingleOutputStreamOperator.addSink(new PrintSinkFunction<>();

這樣就可以了，如果是其他的 Sink Function 的話需要換成對應的。

使用這個 Function 其效果就是列印從 Source 過來的資料，和直接 Source.print() 效果一樣。

下篇文章我們將講解下如何自定義自己的 Sink Function，並使用一個 demo 來教大家，讓大家知道這個套路，且能夠在自己工作中自定義自己需要的 Sink Function，來完成自己的工作需求。

最後

本文主要講了下 Flink 的 Data Sink，並介紹了常見的 Data Sink，也看了下原始碼的 SinkFunction，介紹了一個簡單的 Function 使用, 告訴了大家自定義 Sink Function 的套路，下篇文章帶大家寫個。

關注我

轉載請務必註明原創地址為：http://www.54tianzhisheng.cn/2018/10/29/flink-sink/

另外我自己整理了些 Flink 的學習資料，目前已經全部放到微信公眾號了。你可以加我的微信：zhisheng_tian，然後回覆關鍵字：Flink 即可無條件獲取到。

1、《從0到1學習Flink》—— Apache Flink 介紹

2、《從0到1學習Flink》—— Mac 上搭建 Flink 1.6.0 環境並構建執行簡單程式入門

3、《從0到1學習Flink》—— Flink 配置檔案詳解

4、《從0到1學習Flink》—— Data Source 介紹

5、《從0到1學習Flink》—— 如何自定義 Data Source ？

6、《從0到1學習Flink》—— Data Sink 介紹

7、《從0到1學習Flink》—— 如何自定義 Data Sink ？

8、《從0到1學習Flink》—— Flink Data transformation(轉換)

《從0到1學習Flink》—— Data Sink 介紹