国产伊人一区久久,丁香五月四射,亚洲蜜桃色图一区

image

前言

Data Sources 是什么呢？就字面意思其實就可以知道：數據來源。

Flink 做為一款流式計算框架，它可用來做批處理，即處理靜態(tài)的數據集、歷史的數據集；也可以用來做流處理，即實時的處理些實時數據流，實時的產生數據流結果，只要數據源源不斷的過來，Flink 就能夠一直計算下去，這個 Data Sources 就是數據的來源地。

Flink 中你可以使用 StreamExecutionEnvironment.addSource(sourceFunction) 來為你的程序添加數據來源。

Flink 已經提供了若干實現好了的 source functions，當然你也可以通過實現 SourceFunction 來自定義非并行的 source 或者實現 ParallelSourceFunction 接口或者擴展 RichParallelSourceFunction 來自定義并行的 source，

Flink

StreamExecutionEnvironment 中可以使用以下幾個已實現的 stream sources，

image

總的來說可以分為下面幾大類：

基于集合

1、fromCollection(Collection) - 從 Java 的 Java.util.Collection 創(chuàng)建數據流。集合中的所有元素類型必須相同。

2、fromCollection(Iterator, Class) - 從一個迭代器中創(chuàng)建數據流。Class 指定了該迭代器返回元素的類型。

3、fromElements(T ...) - 從給定的對象序列中創(chuàng)建數據流。所有對象類型必須相同。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Event> input = env.fromElements(
    new Event(1, "barfoo", 1.0),
    new Event(2, "start", 2.0),
    new Event(3, "foobar", 3.0),
    ...
);

4、fromParallelCollection(SplittableIterator, Class) - 從一個迭代器中創(chuàng)建并行數據流。Class 指定了該迭代器返回元素的類型。

5、generateSequence(from, to) - 創(chuàng)建一個生成指定區(qū)間范圍內的數字序列的并行數據流。

基于文件

1、readTextFile(path) - 讀取文本文件，即符合 TextInputFormat 規(guī)范的文件，并將其作為字符串返回。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> text = env.readTextFile("file:///path/to/file");

2、readFile(fileInputFormat, path) - 根據指定的文件輸入格式讀取文件（一次）。

3、readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) - 這是上面兩個方法內部調用的方法。它根據給定的 fileInputFormat 和讀取路徑讀取文件。根據提供的 watchType，這個 source 可以定期（每隔 interval 毫秒）監(jiān)測給定路徑的新數據（FileProcessingMode.PROCESS_CONTINUOUSLY），或者處理一次路徑對應文件的數據并退出（FileProcessingMode.PROCESS_ONCE）。你可以通過 pathFilter 進一步排除掉需要處理的文件。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<MyEvent> stream = env.readFile(
        myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
        FilePathFilter.createDefaultFilter(), typeInfo);

實現:

在具體實現上，Flink 把文件讀取過程分為兩個子任務，即目錄監(jiān)控和數據讀取。每個子任務都由單獨的實體實現。目錄監(jiān)控由單個非并行（并行度為1）的任務執(zhí)行，而數據讀取由并行運行的多個任務執(zhí)行。后者的并行性等于作業(yè)的并行性。單個目錄監(jiān)控任務的作用是掃描目錄（根據 watchType 定期掃描或僅掃描一次），查找要處理的文件并把文件分割成切分片（splits），然后將這些切分片分配給下游 reader。reader 負責讀取數據。每個切分片只能由一個 reader 讀取，但一個 reader 可以逐個讀取多個切分片。

重要注意：

如果 watchType 設置為 FileProcessingMode.PROCESS_CONTINUOUSLY，則當文件被修改時，其內容將被重新處理。這會打破“exactly-once”語義，因為在文件末尾附加數據將導致其所有內容被重新處理。

如果 watchType 設置為 FileProcessingMode.PROCESS_ONCE，則 source 僅掃描路徑一次然后退出，而不等待 reader 完成文件內容的讀取。當然 reader 會繼續(xù)閱讀，直到讀取所有的文件內容。關閉 source 后就不會再有檢查點。這可能導致節(jié)點故障后的恢復速度較慢，因為該作業(yè)將從最后一個檢查點恢復讀取。

基于 Socket：

socketTextStream(String hostname, int port) - 從 socket 讀取。元素可以用分隔符切分。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Tuple2<String, Integer>> dataStream = env
        .socketTextStream("localhost", 9999) // 監(jiān)聽 localhost 的 9999 端口過來的數據
        .flatMap(new Splitter())
        .keyBy(0)
        .timeWindow(Time.seconds(5))
        .sum(1);

這個在《從0到1學習Flink》—— Mac 上搭建 Flink 1.6.0 環(huán)境并構建運行簡單程序入門文章里用的就是基于 Socket 的 Word Count 程序。

自定義：

addSource - 添加一個新的 source function。例如，你可以 addSource(new FlinkKafkaConsumer011<>(...)) 以從 Apache Kafka 讀取數據

說下上面幾種的特點吧：

1、基于集合：有界數據集，更偏向于本地測試用

2、基于文件：適合監(jiān)聽文件修改并讀取其內容

3、基于 Socket：監(jiān)聽主機的 host port，從 Socket 中獲取數據

4、自定義 addSource：大多數的場景數據都是無界的，會源源不斷的過來。比如去消費 Kafka 某個 topic 上的數據，這時候就需要用到這個 addSource，可能因為用的比較多的原因吧，Flink 直接提供了 FlinkKafkaConsumer011 等類可供你直接使用。你可以去看看 FlinkKafkaConsumerBase 這個基礎類，它是 Flink Kafka 消費的最根本的類。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<KafkaEvent> input = env
        .addSource(
            new FlinkKafkaConsumer011<>(
                parameterTool.getRequired("input-topic"), //從參數中獲取傳進來的 topic 
                new KafkaEventSchema(),
                parameterTool.getProperties())
            .assignTimestampsAndWatermarks(new CustomWatermarkExtractor()));

Flink 目前支持如下圖里面常見的 Source：