一、背景
最近在研发实时数据的过程中碰到了需要update写入Explore的大基数实时数据表的场景。为了支持支付宝某个广告位的实时看数需求，需要计算实时累计的流量效果数据。比如曝光pv、点击pv，曝光uv、点击uv，pv曝光点击率。
- 注：Explorer是一款蚂蚁自研海量数据规模下低延时响应的实时分析型(OLAP)数据库, 目标是提供聚合查询能力和一些高阶分析功能。对标业界ClickHouse和阿里的云原生数据仓库AnalyticDB。
现根据业务需求，需要将用户多次行为记录按特定去重规则合并成1条计算。因此设计了一张在user_id、request_id粒度上的大基数explorer表，目标写入数据量在50～80万行/min左右。希望能在此基础上计算出准确的实时累计曝光pv数据。
Blink任务上线后，sink节点数据写入量在30w行/min以上时会持续报错explorer链接失败，任务频繁失败重启，导致延迟持续上涨。本文总结了常用的优化思路和操作，供参考。
com.alibaba.blink.streaming.connectors.common.exception.BlinkRuntimeException: ************ ERR_ID: CON-02010602 CAUSE: Explorer operation failed, msg: Cannot get a connection, pool error Timeout waiting for idle object ACTION: Flush to explorer table failed, contact blink/explorer admin for help. DETAIL: ************
explorer表例子：
二、问题出现的直接原因
直接原因是explorer集群中负载不均匀。部分机器负载过高，大批量写入的时候cpu使用率很高导致写入响应慢，最终造成写入超时报错，Blink任务失败重启。比如，explorer表配置的shard数是10（hashBucket） * 2（shardConfig_task_replicants） = 20，假如explorer集群的机器只有14台，理论上讲会有6台机器的负载比其他机器高，大批量写的时候cpu使用率会很高，写入响应慢， 导致Blink写入超时报错，任务失败重启。
三、优化思路
3.1. 均匀机器负载：通过优化explorer表分片数
- 表的分片数需要根据集群有多少机器进行调整。比如集群机器有14台，可以通过设置hash_bucket=14，让链接都均匀的分布在14台机器上, 不让部分机器负载过高。
{ "shardConfig_partition_columns": "test_column", // 根据该列进行哈希运算 "hash_bucket": "14", // 哈希分桶个数 "update_model": "Row", // 更新模式：追加写入/覆盖写入 "shardConfig_task_replicants": "2", -- 副本数量 "storage_engine": "test_engine", "storage_explorer_tier": "test", }
3.2. 调大超时时间配置
（最简单粗暴的方式，不能从本质上解决问题）
blink是通过jdbc链接的explorer，创建explorer结果表示例如下：
create table explorer_output( user_id varchar, request_id varchar, ..., primary key(rowkey) ) WITH ( -- 写入explorer时的各种参数配置 `user`='test_name' ,`url`='jdbc:mysql:///${test_ip}/cheetah?characterEncoding=utf8&autoReconnect=true&connectTimeout=10000&socketTimeout=30000&rewriteBatchedStatements=true'' ,`zdalpassword`='${test_password}' ,`tablename`='test_table' ,`type`='explorer' ,`cache`='ALL' ,`batchInsertSize`='20000' ,`partitionBy`='rowkey' )
超时配置注意事项：
- blink链接explorer的超时时间由url中connectTimeout和socketTimeout配置。connectTimeout 是blink与explorer TCP建联的超时时间，socketTimeout是blink到explorer TCP读写数据的超时时间。
经尝试，实际将socketTimeOut适当调大后，报错的频率会减少一些。
3.3. 减少写入数据量
3.3.1. Blink sql逻辑优化：通过去重减少输出到sink算子的量
- 【方法1】having count（*） = 1， 使得同维度下有多条数据时，只取第一条。效果类似first_value() OVER partion by xxx order by窗口函数。但是经实测后发现有问题，会丢失数据。怀疑是开了miniBatch微批处理后第一次count就超过1，数据被过滤掉了。
SELECT `user_id`, `request_id`, ... FROM `expo_detail` WHERE `request_id` IS NOT NULL GROUP BY `user_id`, `request_id`, ... having count(*) = 1
- 【方法2】后续改成row_number 方案，相同维度根据日志时间排序取第一条记录即可，示例代码如下：
SELECT `user_id`, `behavior`, `request_id`, ... ROW_NUMBER() OVER ( PARTITION BY `behavior`, `request_id`, ... ORDER BY loged_time -- 顺序排就行 ) AS rn FROM ( SELECT `user_id`, `behavior`, `request_id`, ... FROM `expo_detail` WHERE `request_id` IS NOT NULL ) ) WHERE rn <= 1 -- 只取第一条记录，不丢失即可
3.3.2. Blink sql逻辑优化：过滤冗余数据不参与计算
- 在读取上游数据时可以尽量过滤掉冗余数据，不参与后续算子的计算。
- 本案例中的日志时间有两个，客户端事实发生的日志时间和服务端上报的日志时间。
- 客户端日志时间会存在乱序的情况（比如几天前的数据延迟到达、由于时区不同导致的“未来”的时间）。通过限制log_time（客户端日志时间）在当天内且<=当前最新时间，可以过滤掉不需要的数据，减少一定数据量。
- loged_time服务端上报时间相比客户端日志时间，乱序的情况会少一些。此处由于业务需要，以客户端日志时间为准。
3.3.3. Blink参数优化：限制读取日志流的tps
- 可通过限制上游输入的tps，让数据稳定快速的被处理完输出出去。
- 如果设置过大，在tps高峰出现时，source节点输入tps量会暴涨，给任务带来较大的性能压力，最终也会影响sink节点写入explorer的稳定性。
CREATE TABLE test_table ( .... ) WITH ( -- blink参数配置 `batchGetSize`='5', -- 适当调小，缓解tps高峰时带来的性能压力 ... );
3.3.4. Blink执行计划优化：减少source节点并发
- 可以通过减少source节点的并发，减少下游算子压力
- 如图，可以在Flink UI界面上查看source节点的并发数。一般source节点的并发和上游分片数保持一致。适当调小也可以减少下游写入压力。
3.4. 其他常见延迟调优方法
3.4.1. Blink参数优化：调整explorer写入参数
可以改配置控制Blink写入explorer的频率和一次写入的数据量，提高写入的效率。Explorer写入Blink参数配置说明：
| 参数 | 注释说明 | 备注 |
| batchInsertSize | 一次写入的条数 | 可选，默认200 |
| flushPoolSize | 写入线程数 | 可选，默认1，超过1，写数据会乱序, 如果是update表不能开启 |
| workQueueSize | executor的工作队列大小，和buffer大小成正比，调大就允许更多数据在缓存中 | 可选，默认是20 |
| flushIntervalMs | 刷数据周期，写入速度 1次/2s，30次/m | 默认2s，表示每2s会刷一次数据 |
CREATE TABLE test_table ( `user_id` VARCHAR, `rowkey` VARCHAR, `loged_time` TIMESTAMP, ... primary key(`user_id`, `rowkey`) ) WITH ( -- 部分参数示例 `tablename`='test_table' ,`type`='explorer' ,`cache`='ALL' ,`batchInsertSize`='500' -- 一次写入的条数 默认200 ,`partitionBy`='rowkey' ,`workeQueueSize`='50' -- executor的工作队列大小，默认20 ,`flushIntervalMs`='2000' -- 刷数据周期，默认2s，每2s刷一次数据 );
补充：TaskManager/subTask/slot和线程池/线程和insertBatchSize的关系
一个slot对应一个subTask，一个taskManager假设有32个slot，有32个subTask， 那么一个subTask对应一个线程。整个connectionPool共32个线程，会同时有活跃和非活跃的线程。Sink节点并发数和cpu数会影响subTask的个数。创建explorer结果表的配置insertBatchSize会影响一次写入的数据量。flushInterval影响写入的频率。update表一次写入的数据量增加，耗时会增大。
3.4.2. 避免频繁Full GC
垃圾回收期间，任务会中断执行，影响Blink性能。如何发现Full GC：通过监控Flink metrics可以直接发现。
也可以在Flink UI界面上点击某一个taskManager，点击Metric查看Old Generation GC次数，太大说明存在频繁GC的情况，该taskManager的heap内存不够。
怎么判断内存不够？一般需要缓存大量的数据的地方就需要调大Task Off-Heap堆内存。比如做开窗计算时，需要缓存window窗口段内的数据。补充：
Flink内存模型：https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/deployment/memory/mem_setup_tm/
3.4.3. 其他内存怎么调
- native 内存：开窗计算，聚合计算, 维表关联，去重，这些操作需要调大native 内存。
- Direct memory：当出现DirectBuffer 内存溢出（Out Of Memory）报错时时，可通过修改blink任务参数调大Direct memory。
- 参考：https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/deployment/memory/mem_setup_tm/
四、总结
经过以上方式调优后，在流量正常的情况下，任务不再出现explorer链接失败报错和延迟，写入explorer表的数据量稳定在60-80w行/min。
但从解决指标去重的业务问题的角度来讲，通过存uid、request_id级明细数据的方式计算实时累计曝光pv在这个场景下并不是一个最好的方案。
换个思路，该页面区块的曝光pv需要在uid、request_id级别上去重，看起来是一个多维度去重的问题，但实际上uid和request_id是1对多的关系，不是多对多，对于每一个uid，一个request_id下的多次曝光需要去重，可以转化为直接对request_id去重。计算uv是单维度去重问题的典型例子，参考常见的uv去重的方式，可以采用了hyperLogLog算法（原理见参考文献3）计算曝光pv数据。此处再附一个几种去重方案的性能对比数据，可以看出方案4 explorer存储15min级分时adm数据+hyperLogLog算法计算去重曝光pv比较好（前提是业务看数可以接受1-5%由hyperLogLog带来的误差）。
参考文档/文献
- Flink官方文档：https://nightlies.apache.org/flink/flink-docs-master/zh/docs/concepts/flink-architecture/
- 阿里云-实时计算Blink用户文档： https://help.aliyun.com/apsara/enterprise/v_3_15_0_20210816/sc/enterprise-user-guide/what-is-realtime-compute1.html?spm=a2c4g.14484438.10001.25
- Flajolet, P., Fusy, ?., Gandouet, O., & Meunier, F. (2007). Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm. Discrete mathematics & theoretical computer science, (Proceedings).
作者 | 怀璟
来源 | 阿里云开发者公众号
