引言

希望通過對Flink底層源碼的學習來更深入了解Flink的相關實現(xiàn)邏輯。這里新開一個Flink源碼解析的系列來深入介紹底層源碼邏輯。說明：這里默認相關讀者具備Flink相關基礎知識和開發(fā)經(jīng)驗，所以不會過多介紹相關的基礎概念相關內(nèi)容，F(xiàn)link使用的版本為1.15.2。初步確定按如下幾個大的方面來介紹

計算模型

• DataStream基礎框架

部署&調(diào)度

存儲體系

底層支撐

概覽

本篇是第一篇，介紹計算模型的基礎DataStream的相關內(nèi)容，這一篇只介紹DataStream的基礎內(nèi)容，如如何實現(xiàn)相關的操作，數(shù)據(jù)結構等，不會涉及到窗口、事件事件和狀態(tài)等信息

DataStream是對數(shù)據(jù)流的一個抽象，其提供了豐富的操作算子(例如過濾、map、join、聚合、定義窗口等)來對數(shù)據(jù)流進行處理，下圖描述了Flink中源數(shù)據(jù)通過DataStream的轉換最后輸出的整個過程。

通過上圖可以來構想下，一般一個DataStream具有如下主要屬性

最終整個數(shù)據(jù)流會生成一個DAG圖(有向無環(huán)圖)，通過這個DAG圖就可以生成對應的任務來運行了。下面來具體分析DataStream的實現(xiàn)和生成DAG圖(Flink中叫Graph)

深入DataStream

首先我們通過下圖來看看，DataStream中的一些主要的輔助類，DataStream類本身主要邏輯是對各類轉換關系和sink的操作，而前面說到的一些主要屬性信息都是通過輔助類來處理的。

Transformation:本身主要管理了輸出格式、上游依賴、并行度、id編號等信息以及StreamOperator的工廠類(StreamOperatorFactory)

StreamOperator:主要是各類操作的具體處理邏輯

Function:用戶自定義函數(shù)的接口，如DataStream中map處理時需要傳入的MapFunction就是Function的子接口

DataStream

屬性和方法

DataStream的屬性比較簡單，就2個，1個是實行的環(huán)境信息，另一個是Transformation。
DataStream中的方法主要分為以下幾類

• 基礎屬性信息：如獲取并行度，id，輸出格式等,大多數(shù)是代理來調(diào)用Transformation中對應的方法
• 轉換操作：各類的轉換處理，如map、filter、shuffle、join等
• 輸出處理：各類輸出的sink處理，如保存為文本等，不過大多數(shù)方法都不推薦使用了，這里主要的方法是addSink()
• 觸發(fā)執(zhí)行: 如executeAndCollect，內(nèi)部是調(diào)用了env.executeAsync來執(zhí)行streaming dataflow

除了轉換操作外，其他幾類的邏輯都比較直觀和簡單，這里重點介紹下轉換操作的處理，轉換操作這里分為3類，1.返回是一個DataStream。如map、filter、union、keyBy等；2.返回的是一個Streams，即輸入是多個DataStream，這類的操作主要是多流關聯(lián)的操作，如join、coGroup。這些Streams的類中實現(xiàn)了一些方法，來返回一個DataStream; 3.window類，返回的是AllWindowedStream類型,同樣這些類中也是有方法，來返回一個DataStream。

說明:如上的各個分類都是個人基于理解上做的各個分類處理，非官方定義

類體系

DataStream的類圖關系比較簡單，就如下這幾個類，具體每個子類的信息見下表

Transformation

屬性和方法

DataStream類本身主要是提供了給外部的編程接口的支持，而對Streaming flow算子節(jié)點本身的一些屬性和操作則由Transformation來負責

從上圖可以看出其主要屬性有節(jié)點id，名稱，并行度，輸出類型還有一些與資源相關的內(nèi)容，還有一個是上游的輸入Transformation,由于這個因不同的Transformation會有不同的數(shù)據(jù)個數(shù)，所以這個信息是放在各個子類中的。如ReduceTransformation是有一個input屬性來記錄上游依賴，而如TwoInputTransformation則是有2個屬性input1和input2來記錄上游依賴，另外如SourceTransformation，這個是源頭的Transformation，是沒有上游依賴的Transformation，所以沒有屬性來記錄，但是有個Source屬性來記錄Source輸入
具體對數(shù)據(jù)的操作處理，在Transformation里面有個StreamOperatorFactory屬性，其中的StreamOperator實現(xiàn)了各種的處理算子。注意這里不是所有的Transformation都包含StreamOperatorFactory,如SourceTransformation中就沒有，這個具體大家可以看看相關的代碼。

Transformation的方法基本上是對上述屬性的get和set操作，這里重點要說明一下的是PhysicalTransformation(下面類體系來介紹)中的setChainingStrategy方法，這里的ChainingStrategy是一個枚舉類，主要是控制多個連續(xù)的算子是否可以進行鏈式處理，這個具體的我們在下面介紹StreamGraph時再介紹

類體系

Transformation的大多數(shù)類均為PhysicalTransformation的子類，PhysicalTransformation為有物理操作的，重點是這類的子類是支持Chaining操作的。我們先來看看其重要子類

StreamOperator

屬性和方法

StreamOperator負責對數(shù)據(jù)進行處理的具體邏輯，如map處理的StreamMap，由于各個Operator的處理方式的不同，這里主要以AbstractStreamOperator來介紹一些主要的屬性，如output的數(shù)據(jù)，StreamConfig，StreamingRuntimeContext等。

下面我們重點介紹下相關的方法

StreamOperator接口有定義了重要的3個方法(這里只介紹與數(shù)據(jù)基礎處理相關的部分)

另外關注的對數(shù)據(jù)進行實際處理的方法，

類體系

StreamOpterator的子類非常多，包括測試類的一起有287個，這些大致可以歸屬到如下3個子類中，

Function

Function是針對所有的用戶自定義的函數(shù)，各子類主要是實現(xiàn)對應的，這里定義了種類豐富的各類Function的子接口類來適配各種不同的加工場景，具體的就看源碼了，這里就不詳細介紹了

本節(jié)的最后，我們通過一個例子來看看這幾個類是怎么組合的。如下是一個常見的對DataStream進行map處理的操作

      text
        .flatMap(new Tokenizer)

處理后對應的DataStream的結構如下圖

DataStream生成提交執(zhí)行的Graph

前面分析了DataStream，是單個節(jié)點的，接下來看看整個streaming flow在flink中是怎么轉換為可以執(zhí)行的邏輯的。一般整個數(shù)據(jù)流我們叫做DAG，那在Flink中叫PipeLine，其實現(xiàn)類是StreamGraph。這里先介紹2個概念

• StreamNode：streaming流中的一個節(jié)點，代表對應的算子
• StreamEdge：Graph中的邊，來連接上下游的StreamNode
  
  如上圖所示，圓形為StreamNode,箭頭為StreamEdge,這樣通過這2者就可以構建一個StreamGraph了。
  StreamGraph是最原始的Graph，而其中會做一些優(yōu)化生成JobGraph，最后會生成待執(zhí)行的ExecutionGraph，這里我們先介紹下基礎概念，后面會深入介紹相關的內(nèi)容。
• JobGraph: 優(yōu)化后的StreamGraph，具體做的優(yōu)化就是把相連的算子，如果支持chaining的，合并到一個StreamNode;
• ExecutionGraph: 和JobGraph結構一致

StreamGraph

下面我們來看看StreamGraph的主要屬性和方法，以及如何從DataStream轉換為StreamGraph的。

屬性和方法

重要屬性如下(這里只介紹與生成圖相關的屬性，還有一些如狀態(tài)，存儲類的后面介紹)

說明：StreamGraph只記錄了StreamNode的信息，StreamEdge的信息是記錄在StreamNode中的。如下2個屬性記錄了StreamNode的輸入Edge和輸出Edge

    //StreamNode.java
    private List<StreamEdge> inEdges = new ArrayList<StreamEdge>();
    private List<StreamEdge> outEdges = new ArrayList<StreamEdge>();

主要方法

StreamGraph生成

下面我們來看看DataStream是如何生成StreamGraph的。通過前面對DataStream的分析可知，DataStream的前后依賴關系是通過Transformation來存儲的，這里StreamExecutionEnvironment有個transformations記錄了所有的Transformation

    //StreamExecutionEnvironment.java
    List<Transformation<?>> transformations

這里的數(shù)據(jù)是在DataStream進行轉換處理生成了新的Transformation，同時會把該實例添加到transformations里面，使用的是如下方法

    getExecutionEnvironment().addOperator(resultTransform);

而具體轉換為StreamGraph是通過StreamExecutionEnvironment的 getStreamGraph()方法。最終轉換的邏輯是通過StreamGraphGenerator類來實現(xiàn)。

這里要介紹一個新的類體系TransformationTranslator，有各種的子類來轉換對應類型的Transformation。這里有定義了2個方法分別支持轉換Streaming和Batch。

 //TransformationTranslator.java
 Collection<Integer> translateForBatch(final T transformation, final Context context);
 Collection<Integer> translateForStreaming(final T transformation, final Context context);

對應的映射關系存儲在StreamGraphGenerator類的translatorMap中。

//StreamGraphGenerator.java 
private static final Map<
                    Class<? extends Transformation>,
                    TransformationTranslator<?, ? extends Transformation>>
            translatorMap;
static {
        @SuppressWarnings("rawtypes")
        Map<Class<? extends Transformation>, TransformationTranslator<?, ? extends Transformation>>
                tmp = new HashMap<>();
        tmp.put(OneInputTransformation.class, new OneInputTransformationTranslator<>());
        tmp.put(TwoInputTransformation.class, new TwoInputTransformationTranslator<>());
        tmp.put(MultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
        tmp.put(KeyedMultipleInputTransformation.class, new MultiInputTransformationTranslator<>());
        tmp.put(SourceTransformation.class, new SourceTransformationTranslator<>());
        tmp.put(SinkTransformation.class, new SinkTransformationTranslator<>());
    ...

下面我們通過OneInputTransformationTranslator為例來看看是如何進行轉換的。具體邏輯如下

 //調(diào)用addOperator添加StreamNode
 streamGraph.addOperator(
                transformationId,
                slotSharingGroup,
                transformation.getCoLocationGroupKey(),
                operatorFactory,
                inputType,
                transformation.getOutputType(),
                transformation.getName());
//獲取上游依賴的transformations,然后添加邊
        final List<Transformation<?>> parentTransformations = transformation.getInputs();
        for (Integer inputId : context.getStreamNodeIds(parentTransformations.get(0))) {
            streamGraph.addEdge(inputId, transformationId, 0);
        }

除了添加節(jié)點和邊外，還有一些如設置節(jié)點的并行度等操作，這塊大家可以去看看具體的代碼。
這樣當把所有的Transformation都轉換完，這樣StreamGraph就生成好了。

JobGraph

有了StreamGraph，為什么還需要一個JobGraph呢，這個和Spark中的Stage類似，如果有多個算子能夠合并到一起處理，那這樣性能可以提高很多。所以這里 根據(jù)一定的規(guī)則進行，先我們介紹相關的類

• JobVertex:job的頂點，即對應的計算邏輯(這里用的是Vertex, 而前面用的是Node，有點差異),通過inputs記錄了所有來源的Edge，而輸出是ArrayList來記錄
• JobEdge: job的邊，記錄了源Vertex和咪表Vertex.
• IntermediateDataSet: 定義了一個中間數(shù)據(jù)集，但并沒有存儲，只是記錄了一個Producer(JobVertex)和一個Consumer(JobEdge)
  主要的概念就這些，下面我們看看JobGraph的結構以及如何從StreamGraph轉換為JobGraph

屬性和方法

JobGraph的屬性主要是通過Map<JobVertexID, JobVertex> taskVertices記錄了JobVertex的信息。
另外這個JobGraph是提交到集群去執(zhí)行的，所以會有一些執(zhí)行相關的信息，相關的如下：

    private JobID jobID;
    private final String jobName;
    private SerializedValue<ExecutionConfig> serializedExecutionConfig;
    /** Set of JAR files required to run this job. */
    private final List<Path> userJars = new ArrayList<Path>();
     /** Set of custom files required to run this job. */
    private final Map<String, DistributedCache.DistributedCacheEntry> userArtifacts =
            new HashMap<>();
    /** Set of blob keys identifying the JAR files required to run this job. */
    private final List<PermanentBlobKey> userJarBlobKeys = new ArrayList<>();
    /** List of classpaths required to run this job. */
    private List<URL> classpaths = Collections.emptyList();

而相關的方法主要是

而如何從StreamGraph轉換到JobGraph這塊的內(nèi)容還是比較多，這塊后續(xù)我們單獨開一篇來介紹

總結

本篇從0開始介紹了DataStream的相關內(nèi)容，并深入介紹了DataStream、Transformation、StreamOperator和Function之間的關系。另外介紹了streaming flow轉換為提交執(zhí)行的StreamGraph的過程及StreamGraph的存儲結構。而從StreamGraph->JobGraph->ExecutionGraph這塊涉及的內(nèi)容也較多，且還涉及到提交部署的內(nèi)容，這塊后面單獨來介紹。最后本篇介紹的DataStream只是介紹了最基礎的計算框架，沒有涉及到flink的streaming flow中的時間、狀態(tài)、window等內(nèi)容，更多關于Flink DataStream基礎的資料請關注腳本之家其它相關文章！

最新評論