Flink DataStream 編程入門

流處理是 Flink 的核心，流處理的數據集用 DataStream 表示。數據流從能夠從各類各樣的數據源中建立（消息隊列、Socket 和 文件等），通過 DataStream 的各類 transform 操做，最終輸出文件或者標準輸出。這個過程跟以前文章中介紹的 Flink 程序基本骨架同樣。本篇介紹 DataStream 相關的入門知識。

Flink 101

爲了學習 Flink 的朋友能查看到每一個例子的源碼，我建立了一個 GitHub 項目：https://github.com/duma-repo/awesome-flink 這裏會存放每一篇文章比較重要的示例的源碼，目前支持 Java 和 Scala，仍在不斷完善中。代碼下載後能夠在本地運行，也能夠打包放在集羣上運行。同時，歡迎各位將優質的資源提交到項目中。

簡單示例

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class WindowWordCount {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1)); //空格分割後，每一個單詞轉換成 (word, 1) 二元組輸出
            }
        }
    }

}

這個例子跟之間介紹 WordCount 的例子相似，這裏詳細介紹下涉及的 API 和含義

• 數據源：socketTextStream 是從 socket 建立的數據流，可使用 nc -l 9000 建立 socket 客戶端發送數據
• transform：flatMap 將輸入的數據按照空格分割後，扁平化處理（flat即爲扁平的意思）；keyBy 會按照指定的 key 進行分組，這裏就是將單詞做爲 key；timeWindow 指定時間窗口，這裏是 5s 處理一次；sum 是聚合函數，將分組好的單詞個數求和
• 輸出：print 將處理完的數據輸出到標準輸出流中，能夠在控制檯看到輸出的結果。調用 execute 方法提交 Job

Data Source

通過以上的介紹，咱們知道常見的數據源有 socket、消息隊列和文件等。對於常見的數據源 Flink 已經定義好了讀取函數，接下來一一介紹。

基於文件

• readTextFile(path)：讀文本文件，默認是文件類型是 TextInputFormat，而且返回類型是 String
• readFile(fileInputFormat, path)：讀文件，須要指定輸入文件的格式
• readFile(fileInputFormat, path, watchType, interval, typeInfo)：以上兩個方法內部都會調用這個方法，參數說明：
•  fileInputFormat - 輸入文件的類型
•  path - 輸入文件路徑
•  watchType - 取值爲 FileProcessingMode.PROCESS_CONTINUOUSLY 和 FileProcessingMode.PROCESS_ONCE
• FileProcessingMode.PROCESS_CONTINUOUSLY - 當輸入路徑下有文件被修改，整個路徑下內容將會被從新處理
• FileProcessingMode.PROCESS_ONCE - 只掃描一次，便退出。所以這種模式下輸入數據只讀取一次
•  interval - 依賴 watchType 參數，對於 FileProcessingMode.PROCESS_CONTINUOUSLY 每隔固定時間（單位：毫秒）檢測路徑下是否有新數據
•  typeInfo - 返回數據的類型

須要注意，在底層 Flink 將讀文件的過程分爲兩個子任務 —— 文件監控和數據讀取（reader）。監控任務由 1 個 task 實現，而讀取的任務由多個 task 實現，數量與 Job 的並行度相同。監控任務的做用是掃描輸入路徑（週期性或者只掃描一次，取決於 watchType），當數據能夠被處理時，會將數據分割成多個分片，將分片分配給下游的 reader 。一個分片只會被一個 reader 讀取，一個 reader 能夠讀取多個分片。

基於 Socket

• socketTextStream：從 socket 數據流中讀數據

基於 Collection

• fromCollection(Collection)：從 Java.util.Collection 類型的數據中建立輸入流，collection 中的全部元素類型必須相同
• fromCollection(Iterator, Class)：從 iterator (迭代器）中建立輸入流，Class 參數指定從 iterator 中的數據類型
• fromElements(T ...)：從給定的參數中建立輸入流， 全部參數類型必須相同
• fromParallelCollection(SplittableIterator, Class)：從 iterator 中建立並行的輸入流，Class 指定 iterator 中的數據類型
• generateSequence(from, to)：從 from 至 to 之間的數據序列建立並行的數據流

自定義

• addSource：能夠自定義輸入源，經過實現 SourceFunction 接口來自定義非並行的輸入流；也能夠實現 ParallelSourceFunction 接口或集成 RichParallelSourceFunction 類來自定義並行輸入流，固然也能夠定義好的數據源，如：Kafka，addSource(new FlinkKafkaConsumer08<>(...))

DataStream 的 transform 

以前已經介紹了一些 transfrom 函數，如：map、flatMap 和 filter 等。同時還有窗口函數：window、timeWindow 等，聚合函數：sum、reduce 等。更多的 transform 函數以及使用將會單獨寫一篇文章介紹。

Data Sink

Data Sink 即是數據的輸出。同 Data Source 相似， Flink 也內置了一些輸出函數，以下：

 

• writeAsText(path) / TextOutputFormat：將數據做爲 String 類型輸出到指定文件
• writeAsCsv(...) / CsvOutputFormat：將 Tuple 類型輸出到 ',' 分隔的 csv 類型的文件。行和列的分隔符能夠經過參數配置，默認的爲 '\n' 和 ','
• print() / printToErr()：將數據打印到標準輸出流或者標準錯誤流，能夠指定打印的前綴。
• writeUsingOutputFormat() / FileOutputFormat：輸出到 OutputFormat 類型指定的文件，支持對象到字節的轉換。
• writeToSocket：根據 SerializationSchema 將數據輸出到 socket 
• addSink：自定義輸出函數，如：自定義將數據輸出到 Kafka

小結

本篇文章主要介紹了 Flink Streaming 編程的基本骨架。詳細介紹了 Streaming 內置的 Data Source 和 DataSink 。下篇將繼續介紹 Flink Streaming 編程涉及的基本概念。

代碼地址： https://github.com/duma-repo/awesome-flink/blob/master/chapter-2-flink-streaming/2-1-streaming-starter.md

 

歡迎關注公衆號「渡碼」