本文翻譯自官網(wǎng)：Streaming Aggregation

Flink Table Api & SQL 翻譯目錄

SQL 是*廣泛使用的數(shù)據(jù)分析語(yǔ)言。Flink 的 Table API 和 SQL 使用戶能夠以更少的時(shí)間和精力定義高效的流分析應(yīng)用程序。而且，F(xiàn)link Table API 和 SQL 都得到了有效的優(yōu)化，集成了很多查詢優(yōu)化和優(yōu)化算子的實(shí)現(xiàn)。但是，并非所有優(yōu)化都默認(rèn)啟用，因此對(duì)于某些工作負(fù)載，您可以通過(guò)打開(kāi)某些選項(xiàng)來(lái)提高性能。

在這個(gè)頁(yè)面中，我們將介紹一些有用的優(yōu)化選項(xiàng)和流聚合的內(nèi)部原理模板聚合的翻譯，它們?cè)谀承┣闆r下會(huì)帶來(lái)很大的改進(jìn)。

注意：目前只有 Blink planner 支持本頁(yè)提到的優(yōu)化選項(xiàng)。

注意：目前，流聚合優(yōu)化僅支持無(wú)邊界聚合。以后會(huì)支持窗口聚合的優(yōu)化。

默認(rèn)情況下，無(wú)界聚合算子對(duì)輸入的記錄進(jìn)行一條一條的處理，即(1）從狀態(tài)中讀取累加器，(2）將記錄累加/收回到累加器中，(3）將累加器寫(xiě)回狀態(tài)，（4）下一條記錄會(huì)從（1））再次處理。這種處理方式可能會(huì)增加StateBackend的開(kāi)銷（尤其是RocksDB StateBackend）。另外，生產(chǎn)中很常見(jiàn)的數(shù)據(jù)傾斜會(huì)加劇問(wèn)題并使工作更容易受到背壓情況的影響。

小批量聚合

小批量聚合的核心思想是在聚合算子內(nèi)部的緩沖區(qū)中緩存一組輸入。當(dāng)輸入被觸發(fā)進(jìn)行處理時(shí)，每個(gè)鍵只需一個(gè)操作即可訪問(wèn)狀態(tài)。這可以大大減少狀態(tài)開(kāi)銷并獲得更好的吞吐量。但是，這可能會(huì)增加一些延遲，因?yàn)樗鼤?huì)緩沖一些記錄而不是立即處理它們。這是吞吐量和延遲之間的權(quán)衡。

下圖說(shuō)明了小批量聚合如何減少狀態(tài)操作。

默認(rèn)情況下禁用 MiniBatch 優(yōu)化。要啟用此優(yōu)化，您應(yīng)該設(shè)置 table.exec.mini-batch.enabled、table.exec.mini-batch.allow-latency 和 table.exec.mini-batch.size。請(qǐng)參閱配置頁(yè)面了解更多詳細(xì)信息。

以下示例顯示了如何啟用這些選項(xiàng)。

// instantiate table environment
val tEnv: TableEnvironment = ...
// access flink configuration
val configuration = tEnv.getConfig().getConfiguration()
// set low-level key-value options
configuration.setString("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
configuration.setString("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
configuration.setString("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task

本地全局聚合

提出將局部聚合分為兩個(gè)階段來(lái)解決數(shù)據(jù)傾斜的問(wèn)題，即先在上游進(jìn)行局部聚合，然后在下游進(jìn)行全局聚合，類似于MapReduce中的Combine+Reduce模式. 例如，請(qǐng)考慮以下 SQL：

SELECT color, sum(id)
FROM T
GROUP BY color

數(shù)據(jù)流中的記錄可能會(huì)發(fā)生傾斜，因此聚合運(yùn)算符的某些實(shí)例將處理比其他實(shí)例更多的記錄，這可能會(huì)導(dǎo)致熱點(diǎn)。本地聚合可以幫助將一定數(shù)量的具有相同密鑰的輸入累積到單個(gè)累加器中。全局摘要只會(huì)接收減少的累加器，而不是大量的原始輸入。這樣可以大大降低網(wǎng)絡(luò)重組和狀態(tài)訪問(wèn)的成本。每個(gè)本地聚合累積的輸入數(shù)量基于*小批處理間隔。這意味著本地-全局聚合取決于啟用小批量?jī)?yōu)化。

下圖顯示了本地全局聚合如何提高性能。

以下示例顯示如何啟用本地全局聚合。

// instantiate table environment
val tEnv: TableEnvironment = ...
// access flink configuration
val configuration = tEnv.getConfig().getConfiguration()
// set low-level key-value options
configuration.setString("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
configuration.setString("table.exec.mini-batch.size", "5000")
configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation

拆分不同的聚合

局部和全局優(yōu)化可以有效消除常規(guī)聚合的數(shù)據(jù)傾斜，如SUM、COUNT、MAX、MIN、AVG。然而，在處理不同的聚合反應(yīng)時(shí)，其性能并不令人滿意。

例如，如果我們要分析今天有多少獨(dú)立用戶登錄。我們可能有以下查詢：

SELECT day, COUNT(DISTINCT user_id)
FROM T
GROUP BY day

如果distinct key（即user_id）的值是稀疏的，COUNT DISTINCT不能減少記錄。即使啟用了局部和全局優(yōu)化，也無(wú)濟(jì)于事。因?yàn)槔奂悠魅匀话瑤缀跛械脑加涗?，全局聚合將成為瓶頸（大部分重累加器由一個(gè)任務(wù)處理，即同一天）。

這個(gè)優(yōu)化的思路是把不同的聚合（比如COUNT(DISTINCT col)）分成兩個(gè)層次。第一個(gè)聚合由組鍵和其他桶鍵打亂。用于計(jì)算桶鍵 HASH_CODE(distinct_key)% BUCKET_NUM。BUCKET_NUM 默認(rèn)為 1024，可以通過(guò) table.optimizer.distinct-agg.split.bucket-num 選項(xiàng)進(jìn)行配置。第二次聚合由原始組鍵打亂，用于 SUM 聚合來(lái)自不同桶的 COUNT DISTINCT 值。由于相同的唯一鍵只會(huì)在同一個(gè)桶中計(jì)算，因此轉(zhuǎn)換是等效的。桶鍵作為額外的組鍵，分擔(dān)組鍵中熱點(diǎn)的負(fù)擔(dān)。Bucket 關(guān)鍵字使工作具有可擴(kuò)展性，以解決不同聚合中的數(shù)據(jù)傾斜/熱點(diǎn)問(wèn)題。

拆分非重復(fù)聚合后，上述查詢將自動(dòng)改寫(xiě)為以下查詢：

SELECT day, SUM(cnt)
FROM (
    SELECT day, COUNT(DISTINCT user_id) as cnt
    FROM T
    GROUP BY day, MOD(HASH_CODE(user_id), 1024)
)
GROUP BY day

下圖展示了拆分非重復(fù)聚合如何提高性能（假設(shè)顏色代表日期，字母代表user_id）。

注意：以上是*簡(jiǎn)單的例子，可以從這個(gè)優(yōu)化中受益。此外，F(xiàn)link 支持拆分更復(fù)雜的聚合查詢，例如多個(gè)不同key的（如不同的集合COUNT(DISTINCT a)、SUM(DISTINCT b)），以及其他非重復(fù)的聚合工作（例如模板聚合的翻譯，總和、*大值、*小值、計(jì)數(shù)）。

注意：但是，目前，拆分優(yōu)化不支持包含用戶定義的 AggregateFunction 的聚合。

以下示例顯示了如何啟用拆分不同聚合優(yōu)化。

// instantiate table environment
val tEnv: TableEnvironment = ...
tEnv.getConfig         // access high-level configuration
  .getConfiguration    // set low-level key-value options
  .setString("table.optimizer.distinct-agg.split.enabled", "true")  // enable distinct agg split

在不同的聚合上使用 FILTER 修飾符

在某些情況下，用戶可能需要從不同維度計(jì)算 UV（唯一身份訪問(wèn)者）的數(shù)量，例如 Android UV、iPhone UV、Web UV 和總 UV。很多用戶會(huì)選擇CASE WHEN來(lái)支持這個(gè)功能，例如：

SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('android', 'iphone') THEN user_id ELSE NULL END) AS app_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('wap', 'other') THEN user_id ELSE NULL END) AS web_uv
FROM T
GROUP BY day

但是，在這種情況下，建議使用 FILTER 語(yǔ)法而不是 CASE WHEN。因?yàn)镕ILTER更符合SQL標(biāo)準(zhǔn)，會(huì)得到更多的性能提升。FILTER 是聚合函數(shù)中使用的修飾符，用于限制聚合中使用的值。用 FILTER 修飾符替換上面的例子，如下所示：

SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('android', 'iphone')) AS app_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('wap', 'other')) AS web_uv
FROM T
GROUP BY day

Flink SQL 優(yōu)化器可以在同一個(gè)唯一鍵上識(shí)別不同的過(guò)濾參數(shù)。例如，在上面的示例中，所有三個(gè) COUNT DISTINCT 都在 user_id 列上。那么 Flink 可以只使用一個(gè)共享狀態(tài)實(shí)例而不是三個(gè)狀態(tài)實(shí)例來(lái)減少狀態(tài)訪問(wèn)和狀態(tài)大小。在某些工作負(fù)載中，這可以顯著提高性能。