总结Flink Table & SQL 流式聚合中的几个优化。

MiniBatch

LocalGlobal

Split Distinct

Agg With Filter

MiniBatch
MiniBatch优化的核心思想是缓冲输入记录微批处理以减少对状态的访问，进而提升吞吐并减少数据的输出。

以如下场景为例，看下开启MiniBatch聚合前后的差异。

SELECT key, COUNT(1) 
FROM T 
GROUP BY key

由上图可知:

未开启MiniBatch，每来一条数据，均需要Read State => Acc => Write State一次。假设N条数据，需要操作State 2*N次，输出数据N条。

开启MiniBatch后，会先将数据缓存在聚合算子内部的缓冲区中，到一定数量或时间后，再触发计算。假设缓存了N条数据，M个Key，每个Key需要执行一次Read State => Acc => Write State, 即每个Key需要操作State 2次，M个Key，需要操作State 2*M次，向下游输出数据M条。当M比较集中时，可大大减少读写状态的开销并获得更好的吞吐。

MiniBatch相关的参数:

table.exec.mini-batch.enabled: 是否启用MiniBatch优化。默认false。table.exec.mini-batch.allow-latency: 缓冲的最大等待时间。默认-1 ms。table.exec.mini-batch.size: 缓冲的最大记录数。默认-1。

可通过以下示例开启MiniBatch，如下。

TableEnvironment tEnv = ...
Configuration configuration = tEnv.getConfig().getConfiguration();

// 开启MiniBatch 

configuration.setString("table.exec.mini-batch.enabled", "true"); 
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");

注意：MiniBatch当前仅适用于非Window聚合(Flink 1.10.0)。

LocalGlobal
LocalGlobal优化可以用来解决聚合时的数据倾斜问题。其核心思想是，将聚合分为两个阶段执行，先在上游进行局部(本地/Local)聚合，再在下游进行全局(Global)聚合，类似MapReduce的Combine + Reduce，即先进行一个本地Reduce，再进行全局Reduce。

以如下场景为例，看下开启LocalGlobal聚合前后的差异。

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

这里就直接使用官网的一个图了，如下。

由上图可知:

未开启LocalGlobal优化，由于流中的数据倾斜，Key为红色的聚合算子实例需要处理更多的记录，这就导致了热点问题。

开启LocalGlobal优化后，先进行本地聚合，再进行全局聚合。可大大减少GlobalAgg的热点，提高性能。

LocalGlobal相关的参数

LocalGlobal优化需要先开启MiniBatch，依赖于MiniBatch的参数。

table.optimizer.agg-phase-strategy: 聚合策略。默认AUTO，支持参数AUTO、TWO_PHASE(使用LocalGlobal两阶段聚合)、ONE_PHASE(仅使用Global一阶段聚合)。

可通过以下示例开启LocalGlobal，如下。

TableEnvironment tEnv = ...
Configuration configuration = tEnv.getConfig().getConfiguration();// 开启MiniBatch 
configuration.setString("table.exec.mini-batch.enabled", "true"); 
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");

// 开启LocalGlobal

configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");

注意:

LocalGlobal能有效提升如SUM、COUNT、MAX、MIN和AVG等聚合的性能。

开启LocalGlobal需要UDAF实现Merge方法。

Split Distinct
Split Distinct优化可以用来解决COUNT DISTINCT的热点问题。

如下场景，统计一天的UV。

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

如果user_id比较稀疏，即便开启了LocalGlobal优化，收效也并不明显，因为COUNT DISTINCT在Local阶段时，去重率并不高，这就导致在Global阶段仍然存在热点问题。

为了解决这一问题，需要将原始聚合拆分成两层聚合:

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

第一层聚合: 将Distinct Key打散求COUNT DISTINCT。

第二层聚合: 对打散去重后的数据进行SUM汇总。

下图显示了Split Distinct是如何提高这种场景下的性能的。

Split Distinct相关的参数

table.optimizer.distinct-agg.split.enabled: 启用Split Distinct优化。默认false。table.optimizer.distinct-agg.split.bucket-num: Split Distinct优化在第一层聚合中，被打算的bucket数目。默认1024。

可通过以下示例开启Split Distinct，如下。

TableEnvironment tEnv = ...
Configuration configuration = tEnv.getConfig().getConfiguration();// 开启MiniBatch 
configuration.setString("table.exec.mini-batch.enabled", "true"); 
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");

// 开启LocalGlobal
configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");// 开启Split Distinct
configuration.setString("table.optimizer.distinct-agg.split.enabled", "true");

注意:

目前不能在包含UDAF的Flink SQL中使用Split Distinct优化方法。

拆分出来的两个GROUP聚合还可参与LocalGlobal优化。

从FLink1.9.0版本开始，提供了COUNT DISTINCT自动打散功能，不需要手动重写。

Agg With Filter
在某些场景下，可能需要从不同维度来统计UV，如Android中的UV，iPhone中的UV，Web中的UV和总UV，这时，可能会使用如下CASE WHEN语法。

SELECTday,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语法, 目前的Flink SQL优化器可以识别同一唯一键上的不同FILTER参数。如，在上面的示例中，三个COUNT DISTINCT都作用在user_id列上。此时，经过优化器识别后，Flink可以只使用一个共享状态实例，而不是三个状态实例，可减少状态的大小和对状态的访问。
将上边的CASE WHEN替换成FILTER后，如下所示:

SELECTday,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