导读

基于HDFS的传统数据存储方案由于HDFS等存储平台的限制，只能增加文件不能修改文件中的内容。想要实现某条记录的变更，就需要读取对应的文件并进行重写，效率极低，而且容易引起数据不一致和冲突。Delta Lake是基于Apache Spark的下一代数据湖存储引擎，支持Merge命令，可以高效的完成upsert或删除。

1 为什么需要Upserts?

1 迎合GDPR，general data protection regulation《通用数据保护条例》，这个法案规定需要在使用个人敏感信息时，得到数据主体的同意，并提供自由的查看删除机制。

2 传统数据的CDC，change data capture 变更数据捕获：移动端和web端基于传统数据库实现低延迟的微服务，困难是各个数据库的数据如何统一分析，因此数据分析师需要构建各种ETL把数据汇总到统一的数据湖中。同步数据时经常涉及的变更来自于schema的修改、数据的增删改。

3 会话化：在广告分析或者用户行为分析中，基于会话的事件分析需求是很常见的。但是在数据湖中由于数据都是append追加模式，想要不断的更新会话信息还是比较困难的。比如想把同一个会话中的数据汇总成一条，保存在数据湖中。传统方案需要读取所有的数据，把新的事件追加到会话中，重新写入。

4 去重：一般来说数据湖的数据都是追加模式，因此经常需要在读取数据后进行去重。比如通过每天维护一个业务表的全量信息，使用时需要读取全部数据，按照id取最新的数据。

2 Upserts的挑战

数据湖基于文件，一般都是以追加的形式修改数据。当有数据需要改变时，经常需要读取全部内容重新写回到平台。这种方式：

1 低效：读写整个分区或者表更新少量的记录，会导致延迟和消耗。需要手动调优表和查询

2 出错：手工代码可能会出现BUG，多个操作同时处理一个表时可能报错。

3 难以维护：手工代码难以理解，追踪调试，随着使用的时间，更难维护

Delta Lake就是既能解决更新效率低下，又能进行并发读写控制，还支持时间回溯，查看历史任意时间的数据快照。

3 Delta中的Merge

基于Delta可以很轻松的使用Merge命令完成上面的操作，如下所述，当userId相同时，更新address字段；当userId没有出现在原表中时，则进行插入操作。

MERGE INTO users
USING updates
ON users.userId = updates.userId
WHEN MATCHED THEN
      UPDATE SET address = updates.addresses
WHEN NOT MATCHED THEN
      INSERT (userId, address) VALUES (updates.userId, updates.address)

整个Merge的流程大致:

内部会通过待更新的数据DataFrame与原表进行inner join，找到哪些文件存在数据交集。一方面读取找到的文件的全部内容，与待更新的表进行full outer join，并按照规则更新字段，存储到delta的数据目录，并标记为新增；另一方面直接把找到的文件标记为删除。

因此可以做到如下的优点：

1 细粒度，更新的内容以文件粒度进行修改，而不是分区维度；

2 高效，有数据跳过的机制(inner join后没有标记出的文件什么也不做)，不需要全表读取重写

3 事务，基于乐观锁控制并发写入，基于快照机制支持并发读取

总结来说：

在不使用Delta Lake时，需要分析表中哪些分区需要重写；读取对应分区下的所有数据，与更新数据join，重写。

使用Delta Lake时，直接基于merge命令，自动完成数据的更新插入。

4 总结

基于delta lake的merge机制，可以自动完成数据的更新插入。而且这种数据更新插入以文件为最细粒度，效率高；支持事务；支持并发读写。可以很方便的实现批流的整合。

典型的应用场景：使用structured streaming以微批的形式把变更同步到delta lake，后面对接流或者批实现业务计算。

更多系列内容：