Flink SQL 性能调优--MiniBatch提升吞吐率

Flink 在设计和实现流计算算子时，把“面向状态编程”作为第一准则。因为在流计算中，为了保证状态（State）的一致性，需要将状态数据存储在状态后端（StateBackend），由框架来做分布式快照。而目前主要使用的RocksDB状态后端会在每次read和write操作时发生序列化和反序列化操作，甚至是磁盘的 I/O 操作。因此状态的相关操作通常都会成为整个任务的性能瓶颈，状态的数据结构设计以及对状态的每一次访问都需要特别注意。

微批(MiniBatch)的核心思想就是缓存一小批数据，在访问状态状态时，同key 的数据就只需要发生一次状态的操作。当批次内数据的 key 重复率较大时，能显著降低对状态的访问频次，从而大幅提高吞吐。

下图说明了MiniBatch如何减少状态操作。

minibatch_agg

当未开启 MiniBatch 时，Aggregate 的处理模式是每来一条数据，查询一次状态，进行聚合计算，然后写入一次状态。当有 4条数据时，需要操作 2*4 次状态

当开启 MiniBatch 时，对于缓存下来的 N 条数据一起触发，同 key 的数据只会读写状态一次。例如下缓存的 4 条 A 的记录，只会对状态读写各一次。所以当数据的 key 的重复率越大，攒批的大小越大，那么对状态的访问会越少，得到的吞吐量越高。

适用场景

微批处理通过增加延迟换取高吞吐，如果您有超低延迟的要求，不建议开启微批处理。通常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启。

开启方式

MiniBatch默认关闭，开启方式如下：

// instantiate table environment
TableEnvironment tEnv = ...

// access flink configuration
Configuration 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

MiniBatch生效的SQL语句

MiniBatch主要作用于聚合(Group By)语句中，且不带window的场景。

window agg

select count(a) from t group by tumble(ts, interval ’10’ second), b

以10秒翻转窗口和字段b聚合，该场景MiniBatch不生效

group agg

select count(a) from t group by b

以字段a聚合，该场景MiniBatch生效

over agg

select count(a) over (partition by b order by c) from t

over window，该场景MiniBatch不生效

MiniBatch二种聚合优化

Local-Global 聚合

LocalGlobal优化将原先的Aggregate分成Local+Global两阶段聚合，即MapReduce模型中的Combine+Reduce处理模式。

第一阶段在上游节点本地攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。
第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。

LocalGlobal本质上能够靠localAgg的预聚合筛除部分倾斜数据，从而降低globalAgg的热点，提升性能。

可以结合下图及SQL理解LocalGlobal如何解决数据倾斜的问题。

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

local_agg

使用场景:
- LocalGlobal适用于提升如SUM、COUNT、MAX、MIN和AVG等普通聚合的性能，能提高算子吞吐量，也能有效解决常见数据热点问题。

配置:

table.optimizer.agg-phase-strategy开启（默认值已为AUTO开启，所以不用配置）

// instantiate table environment
TableEnvironment tEnv = ...

// access flink configuration
Configuration 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

Partial-Final聚合

Local-Global聚合能针对常见普通聚合有较好的效果（如SUM、COUNT、MAX、MIN和AVG）。但是对于COUNT DISTINCT效果并不明显。COUNT DISTINCT在local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点。

PartialFinal优化会自动打散成两层聚合，增加按distinct key取模的打散层，解决COUNT DISTINCT的热点问题。

可以结合下图及SQL理解Partial-Final如何解决COUNT DISTINCT热点问题。

-- 原始SQL
SELECT day, COUNT(DISTINCT user_id)
FROM T
GROUP BY day

-- PartialFinal优化后SQL
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

distinct_split

使用场景：
- PartialFinal优化方法适用于使用COUNT DISTINCT且聚合节点性能无法满足时。
- PartialFinal优化方法不能在包含UDAF的Flink SQL中使用。

配置:

table.optimizer.distinct-agg.split.enabled开启（默认值已为false，需要设置为true）
table.optimizer.distinct-agg.split.bucket-num（默认值1024，可以根据业务数据量和热点情况，设置这个取模值）

// instantiate table environment
TableEnvironment tEnv = ...

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

建议：数据量不大的情况下不建议使用PartialFinal优化方法。PartialFinal优化会自动打散成两层聚合，引入额外的网络Shuffle，在数据量不大的情况下，可能反而会浪费资源。

数据抖动现象

所谓数据抖动问题是指，两层 AGG 时，第一层 AGG 发出的更新消息会拆成两条独立的消息被下游消费，分别是retract 消息和 accumulate 消息。而当第二层 AGG 消费这两条消息时也会发出两条消息。从前端看到就是数据会有抖动的现象。例如下面的例子，统计买家数，这里做了两层打散，第一层先做 UV 统计，第二级做SUM。

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

当第一层count distinct的结果从100上升到101时，它会发出 -100, +101 的两条消息。当第二层的 SUM 会依次收到这两条消息并处理，假设此时 SUM 值是 900，那么在处理 -100 时，会先发出 800 的结果值，然后处理 +101 时，再发出 901 的结果值。从用户端的感受就是买家数从 900 降到了 800 又上升到了 901，我们称之为数据抖动。而理论上买家数只应该只增不减的，所以我们也一直在思考如何解决这个问题。

数据抖动的本质原因是 retract 和 accumulate 消息是一个事务中的两个操作，但是这两个操作的中间结果被用户看到了，也就是传统数据库 ACID 中的隔离性中最弱的 READ UNCOMMITTED 的事务保障。要从根本上解决这个问题的思路是，如何原子地处理 retract & accumulate 的消息。