微信实验平台：全面拥抱湖仓时代

　　作者：bearly
　　大数据领域流批一体和湖仓一体都是技术演进的趋势，技术大一统是每个coder的浪漫。微信实验平台从2022年Q1开始引入iceberg作为湖仓一体解决方案，最早从0。14。1版本开始引入到如今的1。2。2版本的广泛使用，目标是为了优化现有流程，达到更快（计算时效性）及更省（节省更多资源）。
　　对比太过朴素的Hive，兼容性不足的THive，Iceberg湖格式带来了极致的灵活性，给我们业务实现方案带来了新的优雅的解决思路。当前实验平台增量存储预算都已经收敛在Iceberg库上（20PB），会逐步迁移“年久失修”的THive，全面拥抱湖仓格式。
　　前言：
　　大数据领域流批一体和湖仓一体都是技术演进的趋势，技术大一统是每个coder的浪漫。微信实验平台从2022年Q1开始引入iceberg作为湖仓一体解决方案，最早从0。14。1版本开始引入到如今的1。2。2版本的广泛使用，目标是为了优化现有流程，达到更快（计算时效性）及更省（节省更多资源）。
　　湖仓一体的背景不过多介绍，及Lamdba、Kappa等架构相信数据工程coder已经深深理解，直接介绍微信实验平台基于iceberg做的改造，和带来的收益，以及过程中遇到的问题，未来的期望。希望能从实用性上给大家介绍我们的解决方案，便于给大家不同业务视角的启迪和借鉴。本文更多从平台业务角度技术方案层面介绍建设思路及合理性论述，针对实现细节问题可以私微一起交流探讨。
　　背景：
　　微信实验平台简介：
　　提供微信内部各个业务场景（视频号、直播、搜一搜、公众号等）下的各类实验场景的支持，有AB实验，MAB实验，BO实验，Interleaving实验，客户端实验，社交网络实验，双边实验等。
　　资源量级：
　　微信实验平台承载的是全微信所有业务的实验场景下的指标计算及统计推断，业务有效指标个数达到了6w，妥妥的资源使用大户，当前规模：
　　大数据场景计算资源totalcore已经有19w存储资源total有25PB
　　OLAP场景计算资源totalcore为2w存储资源total为5。5PB
　　由于成本及稳定性、中心内业务建设等角度考虑，我们计算资源大多收敛在Gemini（微信云原生大数据平台），天穹Gaia（TEG公司级大数据平台）计算资源做Backup，当平台依赖计算集群有异常，可以进行任务层面的计算集群切换，使我们支持的业务用户影响范围最小化，存储资源则完全依赖于天穹。
　　选型分析：
　　Iceberg、Hudi以及DeltaLake是基本同时期出现的开源表存储格式项目，整体的功能和定位也是基本相同，也一定会前期百花齐放相互借鉴最终走向同质化，网上已经有很多相关对比介绍的文章，这里就不详细比较了。
　　我们选择iceberg作为LakehouseTableFormat的方案的主要原因是：
　　公司内部数平团队专门支持，后备力量充足
　　我们场景更多偏离线和近实时（5min），实时场景要求不高，iceberg的高级feature已满足我们的需求
　　其他例如组件抽象更友好、更通用、pluggable设计，向下支持的文件格式（ParquetOrcAvro）、存储类型（ObjectStorageFileStorage）更多，向上支持的计算引擎（SparkFlinkHiveTrinoImpalaSR）更广泛，这些并不是我们业务方的主要考虑点。只要其与我们平台业务依赖的引擎框架都能足够兼容，生产环境稳定性可控、性能优异、解决方案优雅，即，在兼容我们已有技术架构下，优化现有流程，达到更快（计算时效性）及更省（节省更多资源）的目标。
　　方案：
　　1。基于IcebergTableSchema建设优化
　　实验平台业务不是层级复杂、主题域多样的业务数仓建设场景，而是仅具备超大规模指标计算的单一场景。所以我们的工作重点也不在高效复杂的数仓建设上，而是在于大规模指标计算优化上。
　　实验平台指标计算一般分为两种类型的表，命中表：包含某个实验ID及命中uin信息的物理表，业务表：业务配置的用于指标计算及后续假设检验的物理表，一般命中表及业务表及相关的配置逻辑口径会组成具体的某个指标。
　　有经验的数仓同学基本都清楚，表Schema的建设优化，基本都在于分区、分桶、排序、索引等上。
　　业务表在表Schema的优化上会依赖于业务数仓的建设，依托于业务方的能力，实验平台可控性并不强。命中表的Schema恰恰是平台建设优化的重点，一般某天的某个指标计算会绑定具体的实验ID，很自然的会想到按天作为一级分区，实验ID作为二级分区，可以将可用数据最小化，降低后续指标计算的读表IOshuffleIO。可是难点在于之前依赖于THive的建设，命中表一般至少保留3个月，实验ID更是达到2w个，做笛卡尔积后分区个数可达180w，Thive的元数据体系会受限于单点数据库瓶颈，即，存在OMS（厂版Hivemetastore）RDBS单点问题，经常会因为某个表元数据太多导致整个元数据库OMS负载高，导致Thive不可用，影响THive服务上所有业务方保证的SLA，所以Thive一般无法做到此种分区结构。刚好Iceberg的出现，由于其基于HDFS的独立三层元数据体系，可以将元数据信息的压力从OMS分摊到HDFS上，规避OMS的单点瓶颈问题。
　　总结起来，其实就是利用Iceberg的三层元数据体系带来的灵活性，细化业务表多级分区，规避OMS受限于单点数据库RDBS瓶颈的问题，提升后续计算效率。但要注意的是，要考虑到NameNode的元数据膨胀的问题，单HDFS存储集群一般超过8亿metadatafile（目录文件）则处于高负载，会对HDFS存储集群后续稳定性带来压力。
　　2。MergeintoTimetravel代替传统数仓拉链表
　　微信实验平台的会有命中信息增量变更的场景，即，数仓同学所熟悉的缓慢变化维问题。
　　SlowlyChangingDimensions：维度建模的数据仓库中的概念，在现实世界中，维度的属性并不是静态的，它会随着时间的流失发生缓慢的变化。这种随着时间发生变化的维度我们一般称之为缓慢变化维，并且把处理维度表的历史变化信息的问题称为处理缓慢变化维的问题。
　　拉链表：记录历史数据，记录一个事物从开始一直到当前状态的所有变化的信息。处理缓慢变化维问题的典型方案，拉链表的tableschema实现通常加入属性列starttime，endtime来标识对应维度记录的生效时间生命周期，能够支持方便的分析出历史数据变化情况。
　　一般此类问题的传统解决方案都是基于Hive拉链表来实现的，来减少重复的冗余数据，Hive拉链表虽然可以解决业务问题，但是效率和灵活性都较低。我们引入了高效的数据湖表格式Iceberg来解决相应问题。相比于朴素的Hive增加了很多变化和灵活性。
　　Hive拉链表的方式来减少重复的冗余数据，记录加上starttime，endtime作为生效起止时间，但是此种方式带来的新问题就是每日计算时都需要拉取全部数据读入进MapReduceSpark等计算框架内，将新增数据处理后再写入，需要消耗的计算资源很大，如果数据量特别大也很容易导致集群负载压力过大使任务失败。并且在读取拉链表的时候也需要加过滤条件（wherestarttimeand
　　基于新的数据湖表格式Iceberg来更优雅的处理缓慢变化维问题，对比传统解决该问题基于Hive的拉链表方案的优势。归纳起来主要是通过如下方式实现：
　　（1。使用MergeInto替代InsertOverwrite
　　采用MergeInto进行增量数据批量变更（updateinsertdelete）。是通过重写相关文件，即包含在提交中需要更新的行的数据文件来支持MergeInto，相对比与InsertOverwrite的方式，Iceberg只替换受影响的数据文件来提升运行效率写入效率
　　MERGEINTOicebergcatalog。mmexptlakehouse。mmexptcumufindertUSING（selectfirsthitds，uin，exptid，groupid，bucketsrchitfromicebergcatalog。mmexptlakehouse。mmexptdailyfinder）sONt。uins。uinandt。groupids。groupidWHENMATCHEDANDs。dst。firsthitdsTHENUPDATESETt。firsthitdss。dsWHENNOTMATCHEDTHENINSERT（firsthitds，uin，exptid，groupid，bucketsrchit，extint，extstring）VALUES（s。firsthitds，s。uin，s。exptid，s。groupid，s。bucketsrchit，null，null）；
　　（2。使用TimeTravelSnapshot代替拉链表冗余的记录有效起止时间starttime，endtime属性字段，可以使用timetravelinsqlqueries，比如查询2022120701：21：00的历史状态，可以直接用。
　　timetravelto2022120701：21：00SELECTFROMmmexptlakehouse。tableTIMESTAMPASOF2022120701：21：00；
　　该种方式查询也会比拉链表减少很多处理数据量，不需要做有效时间字段的filter（wherestarttimeand
　　另外由于厂内iceberg老版本还不支持timestampasof等语法，icebergissues270我们给厂内数平同学单独提了issue，在icebergmetadata中加入了customtimestamp结合sqlhint来代替timestampasof方式。后续我们计划应用Iceberg1。2。2带来的BranchingandTagging来去做更优雅的SnapshotTimeTravel。
　　（3。特殊情况处理例如历史数据出错，则可以直接回滚到具体出错前的snapshot，让用户在每次提交的snapshot列表中切换，比如versionrollback，setsnapshotid
　　Rollbacktabledb。sampletosnapshotID1：CALLcatalogname。system。rollbacktosnapshot（db。sample，1）SetsthecurrentsnapshotIDforatable。CALLcatalogname。system。setcurrentsnapshot（db。sample，1）
　　然后数据修正后commit到其后的snapshot中。
　　总结起来，其实就是利用Iceberg的三层元数据体系带来的灵活性，可以解决Hive实现传统拉链表方式下的写入效率低，查询效率低，灵活性低，易用性低等问题。在特定业务超大拉链表的场景中，任务写入及查询效率都带来了指数级的提升。
　　针对此方案我们也申请了专利《一种基于数据湖表格式处理缓慢变化维问题的新方法》专利立项编号：2023010065CN
　　3。流批一体
　　Iceberg使CDC场景做分钟级写入成为可能，可以将Iceberg统一流批Pipeline，作为公共上游，使代码复用，减少数据冗余，并从根源上规避数据不一致等问题。同时我们也希望精简全链路，过多的step会增加数据开发的成本，也会降低全链路的稳定性和可靠性。如上图所示，架构也会更加优雅。
　　收益：
　　在我们的使用实践过程中，发现THive兼容性不足，其中默认的ORC为厂内魔改版本，带来一定的对接使用隐患，比如ClickHouseORC外表无法识别。ORC魔改版本在Spark上的优化，也距离原生组件有些差距。
　　总结起来Iceberg方案的优势，对比太过朴素的Hive，兼容性不足的THive，Iceberg带来的高级Feature：包括ACID粗粒度事务语义，可以避免脏读及下游失败等问题，借助于三层Metadata实现的snapshot、timetravel、schemaevolutionpartitionevolution，rowlevelupsertdelete等feature都带来了极致的灵活性。在业务升级、问题回滚相较于朴素的Hive带来了新的优雅的解决思路。配合异步Autooptimizing服务优化数据存储组织方式（定期compact或进行合理排序和分组），提高查询效率，给我们带来很大收益。
　　我们已经将20PB的历史数据迁移到Iceberg库上，并且后续增量数据默认采用Iceberg作为数据基座。
　　结合社区开源版本优化红利，Spark3。3全面接入（GeminionSparkOteam），带来的增强了AQE（adaptivequeryexecution）能力，增加rowlevelruntimefilter来补充DynamicPartitionpruning等Feature，及Iceberg1。2。2的全面接入，我们从计算性能、存储占用两方面进行了优化的实践，最终效果为，计算上总核时优化69。4，节省约20w核时天，存储空间上优化约100PB，总计折合降本预计约3。8kw年。在降本的同时提升了离线计算的效率。计算任务p99耗时减少70，平均任务耗时减少60。
　　遇到的问题：
　　针对数据开发过程中的业务常见问题数据倾斜问题，小文件问题，随机性问题，我们都有遇到，并有一套解决方式供大家参考
　　1。数据倾斜问题
　　1）分区数据倾斜
　　如上方案一描述的，我们采用实验ID作为二级分区，每个实验的命中流量都是不均匀的，尤其针对一些全流量的holdout实验，就针对写入Iceberg的Stage做了单独的大实验倾斜处理，在写入前的重分布过程中，加入了打散化处理
　　valbucketIdHashUdfudf（（exptid：Long，uin：Long）｛valmaxExptIds：ListBuffer〔Long〕maxExptIdsBroadCast。valueif（maxExptIds。contains（exptid））｛exptid。toString（（uin。hashCode（）Integer。MAXVALUE）50）｝else｛exptid。toString｝｝）valicerbergDfdf。withColumn（bucketid，bucketIdHashUdf（col（exptid），col（uin）））。repartition（partitionNum，col（ds），col（bucketid））。sortWithinPartitions（ds，exptid）
　　2）MergeInto写入倾斜
　　在IcebergTBLPROPERTIES中加入了WriteProperties：
　　write。distributionmoderangeDefinesdistributionofwritedata：none：don’hash：range：rangedistributebypartitionkeyorsortkeyiftablehasanSortOrder
　　2。批写小文件问题
　　相对于实时场景下分钟级commit造成snapshot及datafile膨胀的问题，我们的面对的场景是batch场景，基本为日度例行任务，需要合理配置targetSizeInBytes，及合理控制sparkstage的partitionnumber，来规避batch场景写入iceberg的小文件太多问题，即，每个spark的partition都会写入icebergdatafile，如果写入的icebergdatafilewrite。targetfilesizebytes则直接写对应的文件，如果写入的icebergdatafilewrite。targetfilesizebytes则会拆分多个文件split写入。
　　同时，因为我们的存储资源量级太大，也跟数平运维同学，申请了专属独占的HDFS存储集群，来保证业务稳定性，避免NameNode过载导致文件读写延迟变大或者ConnectFailException等问题，并开通了存储集群grafana监控等权限，提前预知集群健康度对任务的影响。
　　3。随机性问题
　　预期中的Spark任务应该具有幂等性，即，任务多次运行的结果应该完全相同，当出现结果不同的随机性问题时，就很难去回放数据。
　　1）Accumulator带来的随机性问题
　　因为我们的超大任务规模比较大（单任务读写20T），运行过程中因为机器的负载等问题，导致taskfail甚至stagefail很正常，恰恰fail造成了Accumulator的数据运行不一致，sparkdocument上有注明
　　Foraccumulatorupdatesperformedinsideactionsonly，Sparkguaranteesthateachtask’supdatetotheaccumulatorwillonlybeappliedonce，i。e。restartedtaskswillnotupdatethevalue。Intransformations，usersshouldbeawareofthateachtask’supdatemaybeappliedmorethanonceiftasksorjobstagesarereexecuted。
　　Accumulator的更新应该在action算子中，而不应该在transformation算子中，来保证Accumulator的更新只会应用一次。
　　2）Random处理数据倾斜带来的随机性问题
　　处理数据倾斜时，常用的方案为在倾斜key上加入随机数来进行打散，但是这种处理方式在ShuffleFail进行retry时，数据会被不同的task重复fetch，导致引入随机性问题。小任务不太容易出现ShuffleFail的问题，超大任务或者集群负载水位较高时，则更容易触发此类问题，应该用取模或者哈希之类的幂等函数来打散倾斜的Key，规避此类随机性问题。
　　4。其他思考
　　另外关于其他IcebergDataSkipping层面的排序、索引等数据组织上优化的考虑，我们也做了一些思考。
　　如前面提及的，实验平台业务不是层级复杂、主题域多样的业务数仓建设场景，而是仅具备超大规模指标计算的单一场景。所以我们的工作重点也不在高效复杂的数仓建设上，而是在于大规模指标计算优化上。业务上决定了我们没有字段点查的场景，所以并没有使用bloomfilter、bitmapfilter等字段索引feature，仅使用Iceberg默认存储文件级别每列的Min、Max信息，并用于TableScan阶段的文件过滤。ZOrder对我们业务场景收益不大，没有太多的基于某个表的多个常用字段进行filter的dataskipping需求。
　　Iceberg使用过程中Bug及Feature，我们给TEG湖仓同学提的issue，数平同学协助解决，再次感谢TEG湖仓团队。
　　icebergissues270icebergtimetravelmergeinto。增量回溯实现hive拉链数仓
　　icebergissues281timestamp类型in操作没有正确进行分区FilterPushDown的问题。
　　未来规划：
　　1。StarRocksIceberg更好的湖仓融合
　　我们的实时指标计算场景，我们没有复杂的ETLpipeline，主要利用OLAP（StarRocksClickHouse）等SQL表达能力强的引擎，作为实时指标计算的依赖引擎。而非需要FlinkStructuredStreaming等可编程能力强的计算框架。
　　ClickHouse是性能优秀的OLAP引擎，但是Clickhousesql表达及优化能力，普适性不足。为了解决clickhouseshuffle问题及冷数据复用避免转移到cos等对象存储，实现极速的SQLonIceberg，更好的AdHocAnalysis体验。我们后续的实时湖仓方案会采用StarRocks3。xIceberg，便于湖仓融合。
　　StarRocks既能兼容TPCDSBenchmark的语法，在ClickBenchBenchmark（https：benchmark。clickhouse。com）打榜上和Top1的Clickhouse性能极其接近，更注定了StarRocks发展上限很高。
　　2。硬件层面计算框架加速
　　我们的业务更多focus在计算框架层的参数优化，和实现逻辑优化，希望可以优化到极致。从另一个视角思考，硬件层面加速计算过程，对我们的Gemini微信云原生大数据平台依赖的CVM，相同成本下1TSSD云盘替换为8块500G高性能云硬盘做RAID0，因为高性能云硬盘本身就是采用3副本来保证数据高可用，做RAID0后可用性也是完全可以接受，同时IO吞吐可以翻4倍（SSD275MBs，8块云硬盘做RAID0吞吐量能达到1GBs），由于离线计算框架更多使用顺序读写，该方案在实施中，预期收益明显。
　　3。SteamingLakehouse
　　伴随着社区ApachePaimon的孵化，我们也希望流批一体架构变的更简洁，在保证性能的前提下，忽略掉流存储MQ和湖存储TableFormat的差异，可以对外作为一个统一组件供业务使用，追求批流一致性语义，提供实时离线一体化的开发体验。