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

Delta Lake 写数据是其最基本的功能，而且其使用和现有的 Spark 写 Parquet 文件基本一致，在介绍 Delta Lake 实现原理之前先来看看如何使用它，具体使用如下：

df.write.format("delta").save("/data/yangping.wyp/delta/test/") 
  
//数据按照 dt 分区 
df.write.format("delta").partitionBy("dt").save("/data/yangping.wyp/delta/test/") 
  
// 覆盖之前的数据 
df.write.format("delta").mode(SaveMode.Overwrite).save("/data/yangping.wyp/delta/test/") 

大家可以看出，使用写 Delta 数据是非常简单的，这也是 Delte Lake 介绍的 100% 兼容 Spark。

Delta Lake 写数据原理

前面简单了解了如何使用 Delta Lake 来写数据，本小结我们将深入介绍 Delta Lake 是如何保证写数据的基本原理以及如何保证事务性。

得益于 Apache Spark 强大的数据源 API，我们可以很方便的给 Spark 添加任何数据源，Delta Lake 也不例外。Delta Lake 就是使用 DataSource V1 版本的 API 实现的一种新的数据源，我们调用 df.write.format("delta") 其实底层调用的是 org.apache.spark.sql.delta.sources.DeltaDataSource 类。为了简单起见，本文介绍的是 Delta Lake 批量写的实现，实时流写 Delta Lake 本文不涉及，后面有机会再介绍。 Delta Lake 批量写扩展了 org.apache.spark.sql.sources.CreatableRelationProvider 特质，并实现了其中的方法。我们调用上面的写数据方法首先会调用 DeltaDataSource 类的 createRelation 方法，它的具体实现如下：

override def createRelation( 
    sqlContext: SQLContext, 
    mode: SaveMode, 
    parameters: Map[String, String], 
    data: DataFrame): BaseRelation = { 
  
  // 写数据的路径 
  val path = parameters.getOrElse("path", { 
    throw DeltaErrors.pathNotSpecifiedException 
  }) 
  
  // 分区字段 
  val partitionColumns = parameters.get(DeltaSourceUtils.PARTITIONING_COLUMNS_KEY) 
    .map(DeltaDataSource.decodePartitioningColumns) 
    .getOrElse(Nil) 
  
  
  // 事务日志对象 
  val deltaLog = DeltaLog.forTable(sqlContext.sparkSession, path) 
  
  // 真正的写操作过程 
  WriteIntoDelta( 
    deltaLog = deltaLog, 
    mode = mode, 
    new DeltaOptions(parameters, sqlContext.sparkSession.sessionState.conf), 
    partitionColumns = partitionColumns, 
    configuration = Map.empty, 
    data = data).run(sqlContext.sparkSession) 
  
  deltaLog.createRelation() 
} 

其中 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 方法实现如下：

override def run(sparkSession: SparkSession): Seq[Row] = { 
    deltaLog.withNewTransaction { txn => 
      val actions = write(txn, sparkSession) 
      val operation = DeltaOperations.Write(mode, Option(partitionColumns), options.replaceWhere) 
      txn.commit(actions, operation) 
    } 
    Seq.empty 
} 

（编辑：52站长网）

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