Flink 之 Data Sink

首先 Sink 的中文釋義為：

下沉; 下陷; 沉沒; 使下沉; 使沉沒; 倒下; 坐下;

 所以，對應 Data sink 意思有點把數據存儲下來（落庫）的意思；

 

Source  數據源  ---- > Compute  計算 -----> sink 落庫

如上圖，Source 就是數據的來源，中間的 Compute 其實就是 Flink 干的事情，可以做一系列的操作，操作完后就把計算后的數據結果 Sink 到某個地方。（可以是 MySQL、ElasticSearch、Kafka、Cassandra 等）。

這里我說下自己目前做告警這塊就是把 Compute 計算后的結果 Sink 直接告警出來了（發送告警消息到釘釘群、郵件、短信等），這個 sink 的意思也不一定非得說成要把數據存儲到某個地方去。

其實官網用的 Connector 來形容要去的地方更合適，這個 Connector 可以有 MySQL、ElasticSearch、Kafka、Cassandra RabbitMQ 等。

 

Data Source 介紹了 Flink Data Source 有哪些，這里也看看 Flink Data Sink 支持的有哪些：

 

 

 

看下源碼有哪些呢？

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

 

 

從上圖可以看到 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/