flink1.7自定義source實現

flink讀取source data

數據的來源是flink程序從中讀取輸入的地方。我們可以使用StreamExecutionEnvironment.addSource（sourceFunction）將源添加到程序中。
flink附帶大量預先實現好的各種讀取數據源的函數，也可以通過為非並行源去實現SourceFunction接口或者為並行源實現ParallelSourceFunction接口或擴展RichParallelSourceFunction來編寫滿足自己業務需要的定制源。

flink預先實現好數據源

下面有幾個預定義的流源可以從StreamExecutionEnvironment訪問

基於文件

readTextFile（path）: 讀取文本文件，該文件要符合TextInputFormat規范，逐行讀取並作為字符串返回。
readFile（fileInputFormat，path）: 根據指定的文件輸入格式指定讀取文件。
readFile（fileInputFormat，path，watchType，interval，pathFilter，typeInfo）: 這是前兩個方法在內部調用的方法。它根據給定的fileInputFormat讀取路徑中的文件。根據提供的watchType，該源可能會定期監視（每間隔ms）該路徑下來到的新數據（FileProcessingMode.PROCESS_CONTINUOUSLY），或者處理當前路徑中的數據后並退出（FileProcessingMode.PROCESS_ONCE）。使用pathFilter，用戶可以進一步排除文件的處理。

基於套接字

socketTextStream : 從套接字讀取。元素可以用分隔符分隔。

基於集合

fromCollection（Collection） : 從Java Java.util.Collection創建一個數據流。集合中的所有元素必須是相同的類型。
fromCollection（Iterator，Class） :從迭代器創建數據流。該類要指定迭代器返回的元素的數據類型。
fromElements（T ...） :根據給定的對象序列創建數據流。所有對象必須是相同的類型。
fromParallelCollection（SplittableIterator，Class） : 並行地從迭代器創建數據流。該類指定迭代器返回的元素的數據類型。
generateSequence（from，to） : 在給定的區間內並行生成數字序列 。

自定義數據原

package com.intsmaze.flink.streaming.source;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple5;
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 java.util.Random;

/**
* @Description: 自定義數據源的模板
* @Author: intsmaze
* @Date: 2019/1/4
*/ 
public class CustomSource {

	private static final int BOUND = 100;

	public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Tuple2<Integer, Integer>> inputStream= env.addSource(new RandomFibonacciSource());

		inputStream.map(new InputMap()).print();

		env.execute("Intsmaze Custom Source");
	}


	/**
	* @Description: 
	* @Author: intsmaze
	* @Date: 2019/1/5
	*/ 
	private static class RandomFibonacciSource implements SourceFunction<Tuple2<Integer, Integer>> {
		private static final long serialVersionUID = 1L;

		private Random rnd = new Random();

		private volatile boolean isRunning = true;
		private int counter = 0;

		/**
		* @Description: 
		* @Param: 
		* @return: 
		* @Author: intsmaze
		* @Date: 2019/1/5
		*/ 
		@Override
		public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
			while (isRunning && counter < BOUND) {
				int first = rnd.nextInt(BOUND / 2 - 1) + 1;
				int second = rnd.nextInt(BOUND / 2 - 1) + 1;
				ctx.collect(new Tuple2<>(first, second));
				counter++;
				Thread.sleep(50L);
			}
		}

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


	/**
	* @Description: 
	* @Param: 
	* @return: 
	* @Author: intsmaze
	* @Date: 2019/1/5
	*/ 
	public static class InputMap implements MapFunction<Tuple2<Integer, Integer>, Tuple5<Integer, Integer, Integer,
			Integer, Integer>> {
		private static final long serialVersionUID = 1L;

		@Override
		public Tuple5<Integer, Integer, Integer, Integer, Integer> map(Tuple2<Integer, Integer> value) throws
				Exception {
			return new Tuple5<>(value.f0, value.f1, value.f0, value.f1, 0);
		}
	}

}