阿里妹导读
在研发实时数据的过程中碰到了需要update写入Explore的大基数实时数据表的场景。本文记录了经过一系列方式调优后，在流量正常的情况下，任务不再出现explorer链接失败报错和延迟的全过程。
一、背景
最近在研发实时数据的过程中碰到了需要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-02010602CAUSE:Exploreroperation failed, msg: Cannot get a connection, pool error Timeout waiting for idle objectACTION:Flushto 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结果表示例如下：
createtable 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`ISNOTNULLGROUPBY`user_id`,`request_id`, ...havingcount(*) = 1
  • 【方法2】后续改成row_number 方案，相同维度根据日志时间排序取第一条记录即可，示例代码如下：
SELECT`user_id`,`behavior`,`request_id`, ... ROW_NUMBER() OVER (PARTITIONBY`behavior`,`request_id`, ...ORDERBY loged_time -- 顺序排就行 ) AS rnFROM ( SELECT`user_id`,`behavior`,`request_id`, ...FROM`expo_detail`WHERE`request_id`ISNOTNULL ) ) WHERE rn <= 1-- 只取第一条记录，不丢失即可
topN语句参考：https://help.aliyun.com/apsara/enterprise/v_3_15_0_20210816/sc/enterprise-developer-guide/topn-statement.html?spm=a2c4g.14484438.10001.163

3.3.2. Blink sql逻辑优化：过滤冗余数据不参与计算

  • 在读取上游数据时可以尽量过滤掉冗余数据，不参与后续算子的计算。
  • 本案例中的日志时间有两个，客户端事实发生的日志时间和服务端上报的日志时间。
  • 客户端日志时间会存在乱序的情况（比如几天前的数据延迟到达、由于时区不同导致的“未来”的时间）。通过限制log_time（客户端日志时间）在当天内且<=当前最新时间，可以过滤掉不需要的数据，减少一定数据量。
  • loged_time服务端上报时间相比客户端日志时间，乱序的情况会少一些。此处由于业务需要，以客户端日志时间为准。

3.3.3. Blink参数优化：限制读取日志流的tps

  • 可通过限制上游输入的tps，让数据稳定快速的被处理完输出出去。
  • 如果设置过大，在tps高峰出现时，source节点输入tps量会暴涨，给任务带来较大的性能压力，最终也会影响sink节点写入explorer的稳定性。
CREATETABLE test_table ( ....) WITH(-- blink参数配置`batchGetSize`='5', -- 适当调小，缓解tps高峰时带来的性能压力 ...);
参数释义：https://help.aliyun.com/apsara/enterprise/v_3_15_0_20210816/sc/enterprise-developer-guide/create-a-log-service-source-table-1.html?spm=a2c4g.14484438.10001.114

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会刷一次数据
CREATETABLE 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带来的误差）。
参考文档/文献
  1. Flink官方文档：https://nightlies.apache.org/flink/flink-docs-master/zh/docs/concepts/flink-architecture/
  2. 阿里云-实时计算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
  3. Flajolet, P., Fusy, É., Gandouet, O., & Meunier, F. (2007). Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm. Discrete mathematics & theoretical computer science, (Proceedings).
阿里云开发者社区，千万开发者的选择
阿里云开发者社区，百万精品技术内容、千节免费系统课程、丰富的体验场景、活跃的社群活动、行业专家分享交流，欢迎点击【阅读原文】加入我们。
继续阅读
阅读原文
关键词
性能
实时
场景
时候
任务