Delta Lake基础操作和原理

目录

• Delta Lake
  • 特性
  • maven依赖
  • 使用aws s3文件系统快速启动
  • 基础表操作
  • merge操作
  • delta lake更改现有数据的具体过程
  • delta表schema
  • 事务日志
    • delta表文件目录
    • 事务日志的一些疑问
  • 需要避免的操作
  • 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>

使用aws s3文件系统快速启动

spark-shell --packages io.delta:delta-core_2.11:0.5.0,org.apache.hadoop:hadoop-aws:2.7.7 
 --conf spark.delta.logStore.class=org.apache.spark.sql.delta.storage.S3SingleDriverLogStore 
 --conf spark.sql.hive.metastore.version=1.2.1 
 --conf spark.hadoop.fs.s3a.access.key=<access-key> 
 --conf spark.hadoop.fs.s3a.secret.key=<secret-key>

基础表操作

//创建 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里只能有一个。
   • 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写多更新少场景）