数据湖分析之Upsert功能详解

1、数据湖分析之Upsert功能详解技术创新，变革未来 目录传统数仓入库痛点新一代数据湖技术下一步展望 传统数仓数据入库痛点典型场景：业务数据(SQL/NoSQL)全量入库典型T+1方案，数据延迟大全量扫瞄源库，给源库造成压力，同时浪费计算和存储资源。例如使用Sqoop限制：源库信息变更不能同步到数仓，例如：Schema变换 传统数仓数据入库痛点改进的场景：CDC导入到Kafka再导入到HBase, KuduHBase, Kudu无法存储全量数据，需要再次导入到Hive库分析整个流程需要多个组件联动，部署复杂限制：仍然无法解决数据Schema变动情况 腾讯大数据TDW入库腾讯大数据场景：TDBAN

2、K入库消息分拣层分拣消息落地成HDFS文件，定时启动Hive任务进行入库操作数据延迟依赖于定时任务，以及Hive入库Job运行情况无法处理延迟到达数据，延迟到达数据重新入库导致数据丢失 新一代数据湖技术 全新的数据入湖方式CDC 新一代数据湖Upsert技术优势ACID & Snapshot读写分离，数据落地即可用历史信息可追溯Row-level updateCDC增量更新延迟数据修正Table Evolution文件组织方式更灵 活，不再依赖目录 分区支持列增删改 Upsert操作简介Upsert是根据一张源表数据，对目标表的数据进行插入、删除和更新。它的典型SQL语法入下所示：MERGE

3、INTO db_name.target_table AS target_alias USING db_name.source_table AS source_alias ON WHEN MATCHED AND THEN WHEN MATCHED AND THEN WHEN NOT MATCHED AND THEN 其中 = DELETE |UPDATE SET * |UPDATE SET column1 = value1 , column2 = value2 . = INSERT * |INSERT (column1 , column2 .) VALUES (value1 , value2 .

4、) Upsert用例：GDPR数据删除GDPR是欧洲的一个关于用户隐私的法案，简单讲就是用户可以选择被忘记的权力，违规者最高会被罚两千万欧元或者4%的营收。MERGE INTO usersUSING users_to_deleteON users_to_delete.userId = users.userIdWHEN MATCHED THEN DELETE Upsert用例：写入CDC数据MERGE INTO users USING (SELECT userId, latest.address AS address, latest.deleted AS deleted FROM (SELECT

5、 userId, MAX(struct(TIME, address, deleted) AS latest FROM changes GROUP BY userId ) ) latestChangeON latestChange.userId = users.userIdWHEN MATCHED AND latestChange.deleted = TRUE THEN DELETE WHEN MATCHED THEN UPDATE SET address = latestChange.addressWHEN NOT MATCHED AND latestChange.deleted = FALS

6、E THEN INSERT (userId, address) VALUES (userId,address) Upsert API使用def upsert(table: Table, microBatchOutputDF: DataFrame, batchId: Long) table.as(t).merge( microBatchOutputDF.as(s), s.key = t.key).whenMatched().updateAll().whenNotMatched().insertAll().execute()readStream.format(“iceberg/delta”).fo

7、reachBatch(upsert _).outputMode(update).start() Upsert基本原理Upsert是一组Update + Delete + Insert 操作的集合，这个操作集以事务方式提 交给表，它的典型实现方式有两种：Copy On Write和Merge on Read.Copy On Write数据读出到内存，进行行更新后替换原来文件优点是后续读快，且不产生小文件Merge On Read相关更新记录落地成Delta文件，读时进行合并写速度快，会产生小文件问题 Delta Lake UpsertDelta Lake的Upsert实现是Copy On Wri

8、te模式，它使用Spark API 进行一系列操作 生成最终Dataset并进行Overwrite操作。分析Dataset数据，找 出需要更新的分区根据Merge条件构 造目标表DatasetJoin 目标表和源表，根 据条件更新列值，删除 行，或插入值保存Dataset到一 个staging目录以事务方式提交，进行OverwriteRecords to insertRecords to update Records to deleteMerge Into partition1 partition2partition4partition3 Apache Hudi UpsertHudi抽象了Me

9、rge on Read表和Copy On Write表，用户可以根据读写延迟需求 选择任意一种表进行Upsert.Copy On Write表的Upsert Apache Hudi UpsertMerge On Read表的Upsert Apache Iceberg UpsertIceberg和Delta Lake，Hudi不一样，它提供了一套基于Data File的事务操作接口，方便 进行包装实现Copy On Write。目前有两种实现方式：第一种：将Upsert拆成三个操作, delete, update, insert合并成一个事务提交val txn = table.newTransa

10、ction() txn.newDelete.deleteFiles(FilesToDelete).commit() txn.newAppend.appendFiles(filesToAppend).commit() mitTransaction()第二种：将目标表和源表进行join, 然后在内存 根据Match条件进行操作并生成文件，再通过事 务提交delete和insert操作。 Apache Iceberg UpsertIceberg Upsert 的Merge on Read实现和Hudi的Merge on Read比较类似，它定义了两种 Delta文件，基于位置的Position Delete和基于表达式的Equality Delete。同时定义了 Base File和Delta File的序号来标识先后顺序以便进行合并操作。 Upsert 相关优化Copy on write和Merge on Read方式都需要一些优化，例如添加索引支持，这 样可以快速定位需要修改记录所在分区和文件。当然还有一些优化如下：Copy on Write过滤条件尽量写足，避免无意义的写放大Merge on ReadMinor: Delete文件转化