just-JL

阅读思路:

1、源码编译

2、功能如何使用

3、实现原理

4、源码阅读(通读+记录+分析)

源码结构

源码分析

元数据

位置:org.apache.spark.sql.delta.actions下的actions文件

Protocol: 当对协议进行向后不兼容的更改时,用于阻止旧客户端读取或写入日志。在执行任何其他操作之前,readers和writers负责检查它们是否满足最低版本

SetTransaction: 设置给定应用程序的提交版本

AddFile: 往table中添加一个新文件

RemoveFile: 从存储库中逻辑删除给定文件

AddCDCFile: 一个包含CDC数据的变化文件

Metadata: 表的元数据信息

CommitInfo: 保存有关表更改的来源信息

事务

位置:org.apache.spark.sql.delta.OptimisticTransaction

关键方法:

def commit(actions: Seq[Action], op: DeltaOperations.Operation): Long = recordDeltaOperation(
      deltaLog,
      "delta.commit") {
...
}

  

主要分成以下三步:

  • prepareCommit
  • doCommitRetryIteratively
  • postCommit

prepareCommit

提交之前对协议版本、元数据合法性、一次事务中元数据改动次数等actions进行校验,并进行相应的改变,返回最终通过的actions。

doCommitRetryIteratively

存储为hdfs

方法名:doCommitRetryIteratively

入参:

  • attemptVersion: 尝试提交的版本
  • actions:要提交的actions
  • isolationLevel:隔离级别(一共三种,这里只用到SnapshotIsolation或者Serializable),若数据会改变则使用Serializable,否则SnapshotIsolation

逻辑:

  • 尝试第一次提交:doCommit,将actions写入delta log对应的目录中,先创建一个临时文件,再进行重命名,因为hdfs rename是原子操作,不需要加锁,若目标文件已存在,则提交失败,抛出异常
  • 若提交失败,则进行固定次数的重试。在重试提交之前会进行逻辑上的冲突检查(checkForConflicts),若检查通过会返回nextAttemptVersion,再进行提交。
  • 提交成功,则返回当前提交的版本号。

 

重要方法:

方法名:checkForConflicts

入参:

  • checkVersion:上一次提交失败的版本号
  • actions:要提交的actions
  • attemptNumber:尝试次数
  • commitIsolationLevel:SnapshotIsolation\Serializable

逻辑:

  • 获取nextAttemptVersion。执行 deltaLog.update(),这里stalenessAcceptable默认是false,所以需要加锁同步执行 updateInternal,通过currentSnapshot中的checkpointVersion获取对应的chk类型文件及其之后的delta类型文件,从这些文件中筛选出最新完成的chk及其之后的delta文件,处理后返回LogSegment对象;接着创建新的snapshot,并替换掉当前的snapshot;将此时snapshot中的version加1作为nextAttemptVersion。
  • 检查冲突,成功则返回nextAttemptVersion。从上一次提交失败的版本号开始循环直到上一步中获取的nextAttemptVersion,获取在nextAttemptVersion之前的版本对应的要提交的actions,将其分为metadataUpdates、removedFiles、txns、protocol、commitInfo这些action,分别进行并发冲突的判断。详细并发冲突类型见:https://docs.delta.io/latest/concurrency-control.html#conflict-exceptions

postCommit

执行提交后的动作。如果需要进行checkpoint,即提交的版本不等于0并且对chk间隔取余等于0,则进行chk,跟据提交的版本创建snapshot,之后调用 Checkpoints.writeCheckpoint 生成对应的CheckpointMetaData对象,接着写入hdfs,最后,如果开启了过期日志清理,则判断已过期的日志进行删除。

 

调用commit的操作

Table deletes, updates, and merges

deletes

方法名:以 delete(condition: String) 为例

一个问题:如何将delete logical plan转到DeleteCommand

=>delta通过继承Rule来实现,如下图代码。

 

执行DeleteCommand run方法

final override def run(sparkSession: SparkSession): Seq[Row] = {
    recordDeltaOperation(tahoeFileIndex.deltaLog, "delta.dml.delete") {
      val deltaLog = tahoeFileIndex.deltaLog
      deltaLog.assertRemovable()
      deltaLog.withNewTransaction { txn =>
        performDelete(sparkSession, deltaLog, txn)
      }
      // Re-cache all cached plans(including this relation itself, if it's cached) that refer to
      // this data source relation.
      sparkSession.sharedState.cacheManager.recacheByPlan(sparkSession, target)
    }
Seq.empty[Row]

}

  

通过deltaLog.withNewTransaction保证删除的原子性,在事务中执行performDelete方法

 

方法名:deltaLog.withNewTransaction

  • 执行startTransaction。更新当前的snapshot,获取到表最新信息,并返回OptimisticTransaction
  • 将该OptimisticTransaction用ThreadLocal包装起来,保证可见性
  • 将OptimisticTransaction作为入参,调用performDelete方法

 

方法名:performDelete

  • 若删除时没有条件,则返回所有的AddFile;若有条件,则继续根据分区列将条件拆分为元数据的条件(根据分区过滤)和其他条件。
  • 若其他条件是空的,则意味着不需要扫描任何数据文件,直接根据分区筛选出要删除的文件集合;否则,先找出待删除的分区文件,接着保留原始的target(LogicalPlan),替换得到newTarget,只包含待删除的AddFile,然后根据newTarget获取到DataFrame,再通过条件过滤得到需要重写的文件。
  • 若无需要重写的文件,则不需要进行删除,返回空;否则,替换创建新的newTarget,获取到DF,过滤得到不需要删除的DF(updatedDF),重新写入文件(txn.writeFiles(updatedDF),返回所有的RemoveFile和AddFile
  • 如果deleteActions不是空,则进行事务commit

 

updates

方法名:以update(condition: Column, set: Map[String, Column])为例

执行UpdateCommand run方法

final override def run(sparkSession: SparkSession): Seq[Row] = {
    recordDeltaOperation(tahoeFileIndex.deltaLog, "delta.dml.update") {
      val deltaLog = tahoeFileIndex.deltaLog
      deltaLog.assertRemovable()
      deltaLog.withNewTransaction { txn =>
        performUpdate(sparkSession, deltaLog, txn)
      }
      // Re-cache all cached plans(including this relation itself, if it's cached) that refer to
      // this data source relation.
      sparkSession.sharedState.cacheManager.recacheByPlan(sparkSession, target)
    }
    Seq.empty[Row]
  }

  

方法名:performUpdate

  • 获取candidateFiles。根据元数据条件和数据条件过滤得到candidateFiles。
  • 如果candidateFiles为空,则返回Nil;否则如果数据条件为空,则需要更新指定分区文件中的所有行,先获取所有文件路径,根据条件重写文件中的所有row,得到rewrittenFiles(AddFile),将candidateFiles中元素更新为RemoveFile,返回deleteActions和rewrittenFiles合并的结果actions;如果数据条件不为空,根据指定条件找出所有受影响的文件filesToRewrite,如果filesToRewrite为空,则返回Nil,否则删除filesToRewrite中包含的文件,根据条件重写文件中的所有row,得到rewrittenFiles(AddFile),返回deleteActions和rewrittenFiles合并的结果actions
  • 如果actions不为空,则进行事务提交

 

merges

示例如下:

调用的方法为:DeltaMergeBuilder.execute()

最终执行方法:MergeIntoCommand.run()

逻辑:

  • 如果可以合并schema,则更新元数据信息,否则不更新。详见:https://docs.delta.io/latest/delta-update.html#automatic-schema-evolution
  • 如果只进行插入并且此时允许优化,则进入 writeInsertsOnlyWhenNoMatchedClauses,避免重写旧文件,只插入新文件。获取目标输出的对应列的表达式集合,获取源表的DF,根据merge的条件只获取目标表中需要的文件并转化成目标表的DF,将源表DF和目标表DF通过leftanti进行join筛选出需要写入的DF,最后写入新的文件中,返回action集合deltaActions
  • 否则,走正常merge逻辑。1) 先找到需要重写的文件filesToRewrite:通过将源表DF和目标表DF进行inner join得到joinToFindTouchedFiles,从中筛选出需要的列(行id、文件名id)的DF,根据行id进行分组求和,统计求和结果大于1的个数记为multipleMatchCount;如果匹配条件的size大于1且是无条件的删除,则即使multipleMatchCount的值大于0也可以正常计算,否则抛出异常。通过文件路径找到对应的AddFile并返回。2)写入所有的变化。1)获取目标输出列,并由此生成新的LogicalPlan,并为输出列创建别名,从而得到新的目标plan。2)如果只有匹配上的条件且允许调优,则join的类型为rightouter,否则为fullouter;将源表DF与目标DF进行join。3)通过join后的plan获取到rows,对其进行mapPartitions操作,执行JoinedRowProcessor中的processPartition,获取输出的DF,写入delta表,返回action集合newWrittenFiles。4)将filesToRewrite转化为RemoveFile,并与newWrittenFiles组合后返回deltaActions
  • 提交事务。

 

重要方法:processPartition

准备

right outer join:参与 Join 的右表数据都会显示出来,而左表只有关联上的才会显示,否则为null。

full outer join:左右表都会显示,但是如果没有关联时会显示,否则左右表会有一个显示为null。

def processPartition(rowIterator: Iterator[Row]): Iterator[Row] = {
  val targetRowHasNoMatchPred = generatePredicate(targetRowHasNoMatch)
  val sourceRowHasNoMatchPred = generatePredicate(sourceRowHasNoMatch)
  val matchedPreds = matchedConditions.map(generatePredicate)
  val matchedProjs = matchedOutputs.map(generateProjection)
  val notMatchedPreds = notMatchedConditions.map(generatePredicate)
  val notMatchedProjs = notMatchedOutputs.map(generateProjection)
  val noopCopyProj = generateProjection(noopCopyOutput)
  val deleteRowProj = generateProjection(deleteRowOutput)
  val outputProj = UnsafeProjection.create(outputRowEncoder.schema)

  def shouldDeleteRow(row: InternalRow): Boolean =
    row.getBoolean(outputRowEncoder.schema.fields.size)

  def processRow(inputRow: InternalRow): InternalRow = {
    if (targetRowHasNoMatchPred.eval(inputRow)) {
      // Target row did not match any source row, so just copy it to the output
      noopCopyProj.apply(inputRow)
    } else {
      // identify which set of clauses to execute: matched or not-matched ones
      val (predicates, projections, noopAction) = if (sourceRowHasNoMatchPred.eval(inputRow)) {
        // Source row did not match with any target row, so insert the new source row
        (notMatchedPreds, notMatchedProjs, deleteRowProj)
      } else {
        // Source row matched with target row, so update the target row
        (matchedPreds, matchedProjs, noopCopyProj)
      }

      // find (predicate, projection) pair whose predicate satisfies inputRow
      val pair = (predicates zip projections).find {
        case (predicate, _) => predicate.eval(inputRow)
      }

      pair match {
        case Some((_, projections)) => projections.apply(inputRow)
        case None => noopAction.apply(inputRow)
      }
    }
  }

  val toRow = joinedRowEncoder.createSerializer()
  val fromRow = outputRowEncoder.createDeserializer()
  rowIterator
    .map(toRow)
    .map(processRow)
    .filter(!shouldDeleteRow(_))
    .map { notDeletedInternalRow =>
      fromRow(outputProj(notDeletedInternalRow))
    }
}

  

逻辑:(左表为source,右表为target)

  • 遍历join后的分区rows,进入processRow方法。1)如果目标row没有匹配上任何源表的row,即只有右表,左表为null,则直接把输入进行输出;2)当左表有值,右表为null,则是未能匹配上,插入左表中对应的数据;3)当左右表都有时,则说明匹配上了,更新右表的row

join后如下表:

source

target

op

null

have value

直接把输入进行输出

have value

null

notMatched+插入左表中对应的数据

have value

have value

matched+更新右表的row

 

 

 

分类:

技术点:

delta

相关文章: