前言

再上一篇文章中 《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

前面文章 《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 的套路，下篇文章帶大家寫個。

.