Flink 實(shí)踐 | 從 Spark 做批處理到 Flink 做流批一體

?GitHub 地址?

• 可以避免代碼重復(fù)，復(fù)用代碼核心處理邏輯
  
  
  代碼邏輯能完全一致是最好的，但這會(huì)有一定的難度。但整體來講，現(xiàn)在的商業(yè)邏輯越來越長(zhǎng)，越來越復(fù)雜，要求也很多，如果我們使用不同的框架，不同的引擎，用戶每次都要重新寫一遍邏輯，壓力很大并且難以維護(hù)。所以整體來講，盡量避免代碼重復(fù)，幫助用戶復(fù)用代碼邏輯，就顯得尤為重要。

• 流批一體有兩個(gè)方向
  
  
  這兩個(gè)方向要考慮的問題很不一樣，目前 Flink 做 Streaming、Spark 做 Batch 等等一些框架在批處理或流處理上都比較成熟，都已經(jīng)產(chǎn)生了很多的單方面用戶。當(dāng)我們想幫助用戶移到另外一個(gè)方向上時(shí)，比如一些商業(yè)需求，通常會(huì)分成兩類，是先從流處理開始到批處理，還是從批處理開始到流處理。之后介紹的兩個(gè)生產(chǎn)實(shí)踐場(chǎng)景案例，正好對(duì)應(yīng)這兩個(gè)方向。

• 減少維護(hù)工作量
  
  
  避免維護(hù)多套系統(tǒng)，系統(tǒng)之間的差異可能非常大，框架和引擎都不一樣，會(huì)帶來比較多的問題。如果公司內(nèi)部有多條 pipeline ，一個(gè)實(shí)時(shí)一個(gè)離線，會(huì)造成數(shù)據(jù)不一致性，因此會(huì)在數(shù)據(jù)驗(yàn)證、數(shù)據(jù)準(zhǔn)確性查詢、數(shù)據(jù)存儲(chǔ)等方面做很多工作，盡量去維護(hù)數(shù)據(jù)的一致性。

• 學(xué)習(xí)更多
  
  
  框架和引擎很多，商業(yè)邏輯既要跑實(shí)時(shí)，也要跑離線，所以，支持用戶時(shí)需要學(xué)習(xí)很多東西。

后面內(nèi)容主要聚焦在 Flink 做 batch 的效果，F(xiàn)link 和 Spark 的簡(jiǎn)單對(duì)比，以及 LinkedIn 內(nèi)部的一些解決方案。分享兩個(gè)生產(chǎn)上的實(shí)例場(chǎng)景，一個(gè)是在機(jī)器學(xué)習(xí)特征工程生成時(shí)如何做流批一體，另一個(gè)是復(fù)雜的 ETL 數(shù)據(jù)流中如何做流批一體。

3.1 案例 A - 機(jī)器學(xué)習(xí)特征工程

在這個(gè)案例下，我們能提供給用戶一些方案選擇，Imperative API 。看下業(yè)界提供的方案：

• 第一個(gè)選擇是即將要統(tǒng)一化的 Flink DataStream API，此前我們?cè)谧龇桨冈u(píng)估時(shí)也有調(diào)研 Flink DataSet API（deprecated），DataStream API 可以做到統(tǒng)一，并且在流處理和批處理方面的支持都是比較完善的。但缺點(diǎn)是，畢竟是 Imperative API ，可能沒有較多的優(yōu)化，后續(xù)應(yīng)該會(huì)持續(xù)優(yōu)化。可以看下 FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API) [1]? 和 FLIP-134: Batch execution for the DataStream API [2]。

• 第二個(gè)選擇是 Spark Dataset ，也是用戶一個(gè)比較自然的選擇。可以用 Dataset API 做 Streaming，區(qū)別于 ?Flink 的 Dataset 、DataStream API 等物理 API，它是基于 Spark Dataframe SQL engine 做一些 type safety，優(yōu)化程度相對(duì)好一些。可以看下文章 Databricks: Introducing Apache Spark Datasets [3] 和 Spark Structured Streaming Programming Guide: Unsupported-operations [4]。

• 第三個(gè)選擇是 Beam On Spark，它目前主要還是用 RDD runner，目前支持帶 optimization 的 runner 還是有一定難度的。之后會(huì)詳細(xì)說下 Beam 在案例 B 中的一些 ongoing 的工作。可以看下 Beam Documentation - Using the Apache Spark Runner [5] 和 BEAM-8470 Create a new Spark runner based on Spark Structured streaming framework [6]。

• History Server - 支持更豐富的 Metrics 等

• 更好的批處理運(yùn)維工具

3.2 案例 B - 復(fù)雜的 ETL 數(shù)據(jù)流

針對(duì)案例 B，作業(yè)中包括很多 join 和 session window，他們之前也是用 Spark SQL 開發(fā)作業(yè)的。很明顯我 們要從 Declartive API 入手，當(dāng)前提供了 3 種方案：

• 第一個(gè)選擇是 Flink Table API/SQL ，流處理批處理都可以做，同樣的SQL，功能支持很全面，流處理和批處理也都有優(yōu)化。可以看下文章 Alibaba Cloud Blog: What's All Involved with Blink Merging with Apache Flink? [8] 和 FLINK-11439 INSERT INTO flink_sql SELECT * FROM blink_sql [9]。

• 第二個(gè)選擇是 ?Spark DataFrame API/SQL ，也是可以用相同的 interface 做批處理和流處理，但是 Spark 的流處理支持力度還是不夠的。可以看下文章 Databricks Blog: Deep Dive into Spark SQL’s Catalyst Optimizer [10] 和 Databricks Blog: Project Tungsten: Bringing Apache Spark Closer to Bare Metal [11]。

• 第三個(gè)選擇是 Beam Schema Aware API/SQL ，Beam 更多的是物理的 API ，在 Schema Aware API/SQL 上目前都在開展比較早期的工作，暫不考慮。所以，之后的主要分析結(jié)果和經(jīng)驗(yàn)都是從 Flink Table API/SQL 和 Spark DataFrame API/SQL 的之間的對(duì)比得出來的。可以看下文章 Beam Design Document - Schema-Aware PCollections [12] 和 Beam User Guide - Beam SQL overview [13]。

Flink 對(duì) window 的支持明顯比其他引擎要豐富的多，比如 session window，用戶用起來非常方便。我們用戶為了實(shí)現(xiàn) session window ，特意寫了非常多的 UDF ，包括做增量處理，把 session 全部 build 起來，把 record 拿出來做處理等等。現(xiàn)在直接用 session window operator ，省了大量的開發(fā)消耗。同時(shí) group 聚合等 window 操作也都是流批同時(shí)支持的。

// Session Event-time Window.window(Session withGap 10.minutes on $"rowtime" as $"w")
// Session Processing-time Window (assuming a processing-time attribute "proctime").window(Session withGap 10.minutes on $"proctime" as $"w")

// Sliding Event-time Window.window(Slide over 10.minutes every 5.minutes on $"rowtime" as $"w")
// Sliding Processing-time Window (assuming a processing-time attribute "proctime").window(Slide over 10.minutes every 5.minutes on $"proctime" as $"w")
// Sliding Row-count Window (assuming a processing-time attribute "proctime").window(Slide over 10.rows every 5.rows on $"proctime" as $"w")

UDF 是在引擎框架之間遷移時(shí)最大的障礙。如果 UDF 是用 Hive 寫的，那是方便遷移的，因?yàn)椴还苁?Flink 還是 Spark 對(duì) Hive UDF 的支持都是很好的，但如果 UDF 是用 Flink 或者 Spark 寫的，遷移到任何一個(gè)引擎框架，都會(huì)遇到非常大的問題，比如遷移到 Presto 做 OLAP 近實(shí)時(shí)查詢。

為了實(shí)現(xiàn) UDF 的復(fù)用，我們 LinkedIn 在內(nèi)部開發(fā)了一個(gè) transport 項(xiàng)目，已經(jīng)開源至 github [14]?上, 可以看下 LinkedIn 發(fā)表的博客：Transport: Towards Logical Independence Using Translatable Portable UDFs [15]。

transport 給所有引擎框架提供一個(gè)面向用戶的 User API ，提供通用的函數(shù)開發(fā)接口，底下自動(dòng)生成基于不同引擎框架的 UDF ，比如 Presto、Hive、Spark、Flink 等。

用一個(gè)共通的 UDF API 打通所有的引擎框架，能讓用戶復(fù)用自己的業(yè)務(wù)邏輯。用戶可以很容易的上手使用，比如如下用戶開發(fā)一個(gè) MapFromTwoArraysFunction:

public class MapFromTwoArraysFunction extends StdUDF2{
    private StdType _mapType;
    @Override    public List getInputParameterSignatures(){        return ImmutableList.of(            "array[K]",            "array[V]"        );    }
    @Override    public String getOutputParameterSignature(){        return "map(K,V)";    }}@Overridepublic void init(StdFactory stdFactory){    super.init(stdFactory);}@Overridepublic StdMap eval(StdArray a1, StdArray a2){    if(a1.size() != a2.size()) {        return null;    }    StdMap map = getStdFactory().createMap(_mapType);    for(int i = 0; i < a1.size; i++) {        map.put(a1.get(i), a2.get(i));    }    return map;}

處理用戶的 SQL 遷移問題 ，用戶之前是用 Spark SQL 開發(fā)的作業(yè)，之后想使用流批一體，改成 Flink SQL 。目前的引擎框架還是比較多的，LinkedIn 開發(fā)出一個(gè) coral 的解決方案，已在 github [16]?上開源，在 facebook 上也做了一些 talk ，包括和 transport UDF 一起給用戶提供一個(gè)隔離層使用戶可以更好的做到跨引擎的遷移，復(fù)用自己的業(yè)務(wù)邏輯。

看下 coral 的執(zhí)行流程，首先作業(yè)腳本中定義 熟悉的 ASCII SQL 和 table 的屬性等，之后會(huì)生成一個(gè) Coral IR 樹狀結(jié)構(gòu)，最后翻譯成各個(gè)引擎的 physical plan。

In-memory Shuffle，Spark 和 Flink 都支持，比較快，但不支持可擴(kuò)展。

Hash-based Shuffle ，Spark 和 Flink 都支持 ， 相比 In-memory Shuffle ，容錯(cuò)性支持的更好一些，但同樣不支持可擴(kuò)展。

Sort-based Shuffle，對(duì)大的 Shuffle 支持可擴(kuò)展，從磁盤讀上來一點(diǎn)一點(diǎn) Sort match 好再讀回去，在 FLIP-148: Introduce Sort-Based Blocking Shuffle to Flink ?[17] 中也已經(jīng)支持。

External Shuffle Service， 在集群非常繁忙，比如在做動(dòng)態(tài)資源調(diào)度時(shí)，外掛服務(wù)就會(huì)非常重要，對(duì) Shuffle 的性能和資源依賴有更好的隔離，隔離之后就可以更好的去調(diào)度資源。FLINK-11805 A Common External Shuffle Service Framework [18] 目前處于 reopen 狀態(tài)。

Disaggregate Shuffle，大數(shù)據(jù)領(lǐng)域都倡導(dǎo) Cloud Native 云原生，計(jì)算存儲(chǔ)分離在 Shuffle Service 的設(shè)計(jì)上也是要考慮的。FLINK-10653 Introduce Pluggable Shuffle Service Architecture [19] 引入了可插拔的 Shuffle Service 架構(gòu)。

Spark 對(duì) Shuffle Service 做了一個(gè)比較大的提升，這個(gè)工作也是由 LinkedIn 主導(dǎo)的 magnet 項(xiàng)目，形成了一篇名稱為 introducing-magnet 的論文 (Magnet: A scalable and performant shuffle architecture for Apache Spark) [20]，收錄到了 LinkedIn blog 2020 里。magnet 很明顯的提升了磁盤讀寫的效率，從比較小的 random range ，到比較大的順序讀，也會(huì)做一些 merging ，而不是隨意的隨機(jī)讀取 shuffle data ，避免 random IO 的一些問題。

通過 Magent Shuffle Service 緩解了 Shuffle 穩(wěn)定性和可擴(kuò)展性方面的問題。在此之前，我們發(fā)現(xiàn)了很多 Shuffle 方面的問題，比如 Job failure 等等非常高。如果想用 Flink 做批處理，幫助到以前用 Spark 做批處理的用戶，在 Shuffle 上確實(shí)要花更大功夫。

• 在 Shuffle 可用性上，會(huì)采用 best-effort 方式去推 shuffle blocks，忽略一些大的 block ，保證最終的一致性和準(zhǔn)確性；

• 為 shuffle 臨時(shí)數(shù)據(jù)生成一個(gè)副本，確保準(zhǔn)確性。

• LinkedIn 非常認(rèn)可和開心看到 Flink 在流處理和批處理上的明顯優(yōu)勢(shì)，做的更加統(tǒng)一，也在持續(xù)優(yōu)化中。

• Flink 批處理能力有待提升，如 history server，metrics，調(diào)試。用戶在開發(fā)的時(shí)候，需要從用戶社區(qū)看一些解決方案，整個(gè)生態(tài)要搭建起來，用戶才能方便的用起來。

• Flink 需要對(duì) shuffle service 和大集群離線工作流投入更多的精力，確保 workflow 的成功率，如果規(guī)模大起來之后，如何提供更好的用戶支持和對(duì)集群進(jìn)行健康監(jiān)控等。

• 隨著越來越多的框架引擎出現(xiàn)，最好能給到用戶一個(gè)更加統(tǒng)一的 interface，這一塊的挑戰(zhàn)是比較大的，包括開發(fā)和運(yùn)維方面，根據(jù) LinkedIn 的經(jīng)驗(yàn)，還是看到了很多問題的，并不是通過一個(gè)單一的解決方案，就能囊括所有的用戶使用場(chǎng)景，哪怕是一些 function 或者 expression，也很難完全覆蓋到。像 coral、transport UDF。

原視頻：

https://www.bilibili.com/video/BV13a4y1H7XY?p=12

? Flink?Forward?Asia 2021??

???戳我，報(bào)名 FFA 2021 大會(huì)！