Delta Lake写数据使用及实现原理代码解析

mode 就是保持数据的模式，支持 Append、Overwrite、ErrorIfExists 以及 Ignore 等。parameters 这个传递的参数，比如分区字段、数据保存路径以及 Delta 支持的一些参数(replaceWhere、mergeSchema、overwriteSchema 等，具体参见 org.apache.spark.sql.delta.DeltaOptions);data 就是我们需要保存的数据。

createRelation 方法紧接着就是获取数据保存的路径，分区字段等信息。然后初始化 deltaLog，deltaLog 的初始化会做很多事情，比如会读取磁盘所有的事务日志(_delta_log 目录下)，并构建最新事务日志的最新快照，里面可以拿到最新数据的版本。由于 deltaLog 的初始化成本比较高，所以 deltaLog 初始化完之后会缓存到 deltaLogCache 中，这是一个使用 Guava 的 CacheBuilder 类实现的一个缓存，缓存的数据保持一小时，缓存大小可以通过 delta.log.cacheSize 参数进行设置。只要写数据的路径是一样的，就只需要初始化一次 deltaLog，后面直接从缓存中拿即可。除非之前缓存的 deltaLog 被清理了，或者无效才会再次初始化。DeltaLog 类是 Delta Lake 中最重要的类之一，涉及的内容非常多，所以我们会单独使用一篇文章进行介绍。

紧接着初始化 WriteIntoDelta，WriteIntoDelta 扩展自 RunnableCommand，Delta Lake 中的更新、删除、合并都是扩展这个类的。初始化完 WriteIntoDelta 之后，就会调用 run 方法执行真正的写数据操作。WriteIntoDelta 的 run 方法实现如下：

开启事务之前，需要更新当前表事务的快照，因为在执行写数据之前，这张表可能已经被修改了，执行 update 操作之后，就可以拿到当前表的最新版本，紧接着开启乐观事务锁。thunk(txn) 就是需要执行的事务操作，对应 deltaLog.withNewTransaction 里面的所有代码。

我们回到上面的 run 方法。val actions = write(txn, sparkSession) 就是执行写数据的操作，它的实现如下：

（编辑：唐山站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!