Delta Lake基础操作和原理

Delta Lake

特性

• 支持ACID事务
• 可扩展的元数据处理
• 统一的流、批处理API接口
• 更新、删除数据，实时读写（读是读当前的最新快照）
• 数据版本控制，根据需要查看历史数据快照，可回滚数据
• 自动处理schema变化，可修改表结构

maven依赖

<dependency>
  <groupId>io.delta</groupId>
  <artifactId>delta-core_2.11</artifactId>
  <version>0.5.0</version>
</dependency>

因为要方便跑 demo，我这里选择用 spark-shell 来交互式探索一下 Delta Lake 的功能。

按照文档介绍，Delta Lake 是需要 Spark 2.4.2 或以上版本的，所以大家最好去官网下载一个预先编译的 Spark 包。

按照上图，输入命令 bin/spark-shell --packages io.delta:delta-core_2.11:0.5.0 就可以启动加载了 Delta Lake 的 spark shell 了。

关于 --packages 的用法，是因为 Spark 有个专门解析选项参数的工具叫做 SparkSubmitOptionParser，他可以解析到依赖，并且先在本地仓库找，没有的话就会根据你的 Maven 配置到远程拉取，这里 Spark

内部做了一些事情。

如何没有配置maven仓库的话，可以先下载编译好的jar包放到服务器目录然后通过--jars命令加载此jar包即可

bin/spark-shell --jars ./jars/delta-core_2.11-0.5.0-SNAPSHOT.jar 

基础表操作

//创建 delta 表和分区表
val data = spark.range(0, 5) data.write.format("delta").save("/tmp/delta-table") data.write.format("delta").partitionBy("date").save("/tmp/delta-table") //读delta表 //第一种方式
val df = spark.read.format("delta").load("/tmp/delta-table") //第二种方式
val deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table") //覆盖delta表数据(mode换成append就是插入数据) //注意：delta lake会记忆表的schema，默认情况下，overwrite只会更改表数据，不会更改表结构 //注意：可通过.option("mergeSchema", "true")，将df中有而schema没有的字段添加到schema中，也就是add column //注意：在overwrite时，可通过df.write.option("overwriteSchema", "true")来替换原有的schema
val data = spark.range(5, 10) data.write.format("delta").mode("overwrite").save("/tmp/delta-table") //更新delta表数据
deltaTable.update( condition = expr("id % 2 == 0"), set = Map("id" -> expr("id + 100")) ) // Upsert (merge) delta表数据
val newData = spark.range(0, 20).toDF deltaTable.as("oldData") .merge( newData.as("newData"), "oldData.id = newData.id") .whenMatched(col("date") > "2019-01-01") .update(Map("id" -> col("newData.id"))) .whenMatched .delete() .whenNotMatched .insert(Map("id" -> col("newData.id"))) .execute() //读写流式数据到delta表（可以边写边读） //有append和complete两种输出模式，complete和overwrite意思一样
val streamingDf = spark.readStream.format("delta").load("/delta/events") val stream = streamingDf.select("value").as("id").writeStream.format("delta").outputMode("append").option("checkpointLocation", "/tmp/checkpoint").start("/tmp/delta-table") val stream2 = spark.readStream.format("delta").load("/tmp/delta-table").writeStream.outputMode("complete").format("console").start() //删除delta表数据（只是添加墓碑标记，不是物理删除）
deltaTable.delete("id % 2 == 0") deltaTable.delete(condition = expr("id % 2 == 0")) import org.apache.spark.sql.functions._ import spark.implicits._ deltaTable.delete(col("date") < "2017-01-01") //根据时间戳或者版本号查看历史数据（time travel）
val timestamp_string = "2019-01-01" val version = "0" val df1 = spark.read.format("delta").option("timestampAsOf", timestamp_string).load("/delta/events") val df2 = spark.read.format("delta").option("versionAsOf", version).load("/delta/events") //查看数据的历史版本
deltaTable.history().show() spark.sql("DESCRIBE HISTORY '" + pathToEventsTable + "'").show() //只保留最新的数据（vacuum是用来清理磁盘上的历史数据）
deltaTable.vacuum(0) spark.sql(“VACUUM ‘” + pathToEventsTable + “‘ RETAIN 24 HOURS”)

merge操作

merge是delta lake的重要操作，它实现了upsert和delete功能。(示例详见基础表操作)

1. merge最多有两个whenMatched和一个whenNotMatched，且至少有一个when；
2. 如果有两个whenMatched，则第一个whenMatched必须有条件，否则会报错。whenNotMatched可有条件，也可没有；
3. whenMatched最多有一个更新操作和一个删除操作，whenNotMatched最多有一个删除操作，相同操作在一个when里只能有一个。merge的实现就是inner join、left anti join以及full join（详见源码分析）。
   • match了才能删除，不支持删除没有match的数据；
   • match了肯定只能update，没match也没法update，只能insert；
4. merge的实现就是inner join、left anti join以及full join（详见源码分析）。

delta lake更改现有数据的具体过程

delta lake以增量写文件的方式支持数据的更新和删除。

1. 匹配数据，定位需要删除的行和涉及的文件；
2. 将这些文件中需要保留的数据重写到新的文件，然后给旧文件打上墓碑标记。
3. 删除、更新、合并（merge）都是这个流程。

delta表schema

delta lake对schema的验证很严格，但同时也支持schema更改。

1. delta表的schema以json格式保存在事务日志中；（schema属于元数据里的Metadata，存放在commit log的json里，详见源码分析）
2. delta表在写数据的时候，如果dataframe包含了表schema中没有的字段，那么会报错；
3. delta表在写数据的时候，dataframe中的数据类型要和schema中的一致，否则会报错；
4. delta表在写数据的时候，如果dataframe中没有表schema中对应的字段，那么在写数据时，对应字段默认为空值；
5. delta表的schema中，字段名的小写不能相同，如'My'和'my'不能作为两个不同的字段名；（因为虽然delta lake区分大小写，但保存时不敏感，而parquet保存时是大小写敏感的，所以加了这个限制）
6. 可通过.option("mergeSchema", "true")，将df中有而schema没有的字段添加到schema中，也就是add column；
7. 在overwrite时，可通过df.write.option("overwriteSchema", "true")来替换原有的schema。（上面的“基础表操作”里有示例）

事务日志

事务日志是delta lake的核心，它记录了delta表相关的所有commit操作。

1. delta表是一个目录，表的根目录除了表数据外，有一个_delta_log目录，用来存放事务日志；
2. 事务日志记录了从最初的delta表开始的所有commit事件，每个commit形成一个json文件，文件名是严格递增的，文件名就是版本号。
3. （默认）每10个json合并成一个parquet格式的checkpoint文件，记录之前所有的commit。
4. 事务日志有一个最新checkpoint的文件（_delta_log/_last_checkpoint），spark读的时候会自动跳到最新的checkpoint，然后再读之后的json。
5. delta lake 使用乐观的并发控制，当多个用户同时写数据时，（读数据是读当前最新版本的快照，互不影响），都是生成一个新版本的数据文件（文件名不重复），在提交commit时生成下一个版本的日志文件，因为日志版本号是连续递增的，如果检测到了同名的文件已存在，则说明有其他用户执行了新的commit，此时进行冲突检测，如果检测通过，则更新当前的snapshot，然后继续提交commit，如果未通过冲突检测，则报错。
6. 因为事务日志的存在，可以找到历史版本的数据，这也是时间穿梭的实现原理，delta lake可以根据commit记录生成历史版本的数据。
7. 新版本的数据生成后，旧版本的数据不会立刻从磁盘删除，可以使用 VACUUM 命令来删除磁盘上的历史版本数据。

delta表文件目录

delta_table_path/
|-- _delta_log/
|--|-- 00000000000000000000.json
|--|-- 00000000000000000001.json
|--|-- 00000000000000000002.json
|--|-- 00000000000000000003.json
|--|-- 00000000000000000004.json
|--|-- 00000000000000000005.json
|--|-- 00000000000000000006.json
|--|-- 00000000000000000007.json
|--|-- 00000000000000000008.json
|--|-- 00000000000000000009.json
|--|-- 00000000000000000010.json
|--|-- 00000000000000000010.checkpoint.parquet
|--|-- 00000000000000000011.json
|--|-- 00000000000000000012.json
|--|-- _last_checkpoint
|-- part-00000-1339ec93-7d47-4ef7-b167-1e5aaa8cd75d-c000.snappy.parquet
|-- part-00000-10a95e81-d64c-40ff-9143-25e998aadcc5-c000.snappy.parquet
|-- part-00000-22f8124e-d2dd-4804-9037-b7a780f70a08-c000.snappy.parquet
|-- part-00000-7866ec4b-b955-4d86-b08c-58cfc71bc1ea-c000.snappy.parquet
|-- part-00000-d6431884-390d-4837-865c-f6e52f0e2cf5-c000.snappy.parquet

事务日志的一些疑问

Q: 谁来合并json日志？
A: OptimisticTransaction里commit时会调用postCommit()函数，这里会检查日志的版本是否能整除checkpointInterval，如果能则调用deltaLog.checkpoint()函数生成新的checkpoint文件。（详见源码分析）

Q: 生成历史版本的快照是从事务日志里一点点计算得来的？
A: 是的，但无需计算所有日志，只需要计算checkpoint和json文件。可以通过_last_checkpoint直接定位到最新的checkpoint，checkpoint只是单纯的合并日志信息，减少读取文件的数量，并不改变内容。

Q: 最新版本的数据不会删除，删除历史版本后再想用时间穿梭，就得根据commit日志重新计算？
A: 旧版本的数据删除后，就不能用时间穿梭了，时间穿梭只能用于已存在的版本数据。

关于delta lake的事务日志，可以看这篇博客，讲解的很详细
https://mp.weixin.qq.com/s?__biz=MzA5MTc0NTMwNQ==&mid=2650717784&idx=2&sn=53174b4dd05642d0d8746b10555ddcf2&chksm=887da32ebf0a2a38ec4b0e159994c915ee460d554eb84feea993ab0a2f72efc4eb715a07d145&scene=21#wechat_redirect

需要避免的操作

1. 手动更改delta表文件。delta lake是使用事务日志来管理表的信息，即使手动添加了文件，因为事务日志里没有此文件的信息，也读取不了；
   而如果手动删除了文件，事务日志中该文件的指针依然存在，但无法读取。
2. 使用其他文件读取方式。delta lake的数据是按照parquet格式存储的，可以使用各种工具读取，但是使用其他读取方式，数据会存在安全隐患（官档说的，其实只要别改文件内容，读应该没啥影响）。

delta lake目前的不足

• 更新操作很重，更新一条数据和更新一批数据的成本可能是一样的，所以不适合一条条的更新数据
• 更新数据的方式是新增文件，会造成文件数量过多，需要清理历史版本的数据，version最好不要保存太多
• 乐观锁在多用户同时更新时并发能力较差，更适合写少读多的场景（或者only append写多更新少场景）