小编给大家分享一下spark delta如何读数据，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！
分析
spark 的delta datasource的构建要从DataSource.lookupDataSourceV2开始，之后会流向到loadV1Source，这里会进行dataSource.createRelation进行构建datasource的Relation的构建,直接转到deltaDataSource 的createRelation：
override def createRelation( sqlContext: sqlContext, parameters: Map[String, String]): BaseRelation = { val maybePath = parameters.getorElse("path", { throw DeltaErrors.pathNotSpecifiedException }) // Log any invalid options that are being passed in DeltaOptions.verifyOptions(CaseInsensitiveMap(parameters)) val timeTravelByParams = DeltaDataSource.getTimeTravelVersion(parameters) DeltaTableV2( sqlContext.sparkSession, new Path(maybePath), timeTravelOpt = timeTravelByParams).toBaseRelation }
DeltaOptions.verifyOptions进行参数校验,有效的参数如下：
val validOptionKeys : Set[String] = Set( REPLACE_WHERE_OPTION, MERGE_SCHEMA_OPTION, EXCLUDE_REGEX_OPTION, OVERWRITE_SCHEMA_OPTION, USER_MetaDATA_OPTION, MAX_FILES_PER_TRIGGER_OPTION, IGnorE_FILE_DELETION_OPTION, IGnorE_CHANGES_OPTION, IGnorE_DELETES_OPTION, OPTIMIZE_WRITE_OPTION, DATA_CHANGE_OPTION, "queryName", "checkpointLocation", "path", "timestampAsOf", "versionAsOf" )
DeltaDataSource.getTimeTravelVersion根据指定的timestampAsOf或者versionAsOf获取指定的版本
def toBaseRelation: BaseRelation = {
if (deltaLog.snapshot.version == -1) {
val id = catalogTable.map(ct => DeltaTableIdentifier(table = Some(ct.identifier)))
.getorElse(DeltaTableIdentifier(path = Some(path.toString)))
throw DeltaErrors.notADeltaTableException(id)
}
val partitionPredicates = DeltaDataSource.verifyAndCreatePartitionFilters(
path.toString, deltaLog.snapshot, partitionFilters)
// Todo(burak): We should pass in the snapshot here
deltaLog.createRelation(partitionPredicates, timeTravelSpec)
}
如果存在分区，则DeltaDataSource.verifyAndCreatePartitionFilter创建partitionPredicates
timeTravelSpec，这里优先选择用户指定的timeTravelByParams，否则通过DeltaDataSource.parsePathIdentifier选择path指定的version，格式如:/some/path/partition=1@v1234 或者/some/path/partition=1@yyyyMMddHHmmssSSS
-
直接调用deltaLog.createRelation:
def createRelation( partitionFilters: Seq[Expression] = Nil, timeTravel: Option[DeltaTimeTravelSpec] = None): BaseRelation = { val versionToUse = timeTravel.map { tt => val (version, accesstype) = DeltaTableutils.resolveTimeTravelVersion( spark.sessionState.conf, this, tt) val source = tt.creationSource.getorElse("unkNown") recordDeltaEvent(this, s"delta.timeTravel.$source", data = Map( "tabLeversion" -> snapshot.version, "queriedVersion" -> version, "accesstype" -> accesstype )) version } /** Used to link the files present in the table into the query planner. */ val snapshottoUse = versionToUse.map(getSnapshotAt(_)).getorElse(snapshot) val fileIndex = TahoeLogFileIndex( spark, this, dataPath, snapshottoUse.Metadata.schema, partitionFilters, versionToUse) new HadoopFsRelation( fileIndex, partitionSchema = snapshottoUse.Metadata.partitionSchema, dataSchema = snapshottoUse.Metadata.schema, bucketSpec = None, snapshottoUse.fileFormat, snapshottoUse.Metadata.format.options)(spark) with InsertableRelation { def insert(data: DataFrame, overwrite: Boolean): Unit = { val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Append WriteIntoDelta( deltaLog = DeltaLog.this, mode = mode, new DeltaOptions(Map.empty[String, String], spark.sessionState.conf), partitionColumns = Seq.empty, configuration = Map.empty, data = data).run(spark) } }override def inputFiles: Array[String] = { getSnapshot(stalenessAcceptable = false).filesForScan( projection = Nil, partitionFilters).files.map(f => absolutePath(f.path).toString).toArray }该方法调用了snapshot的filesForScan方法：
def filesForScan(projection: Seq[Attribute], filters: Seq[Expression]): DeltaScan = { implicit val enc = SingleAction.addFileEncoder val partitionFilters = filters.flatMap { filter => DeltaTableutils.splitMetadataAndDataPredicates(filter, Metadata.partitionColumns, spark)._1 } val files = DeltaLog.filterFileList( Metadata.partitionSchema, allFiles.toDF(), partitionFilters).as[AddFile].collect() DeltaScan(version = version, files, null, null, null)(null, null, null, null) }
通过之前文章的分析，我们直到deltalog记录了AddFile和Remove记录，那现在读数据怎么读取呢？全部在allFiles方法。
重点看一下：allFiles方法：
def allFiles: Dataset[AddFile] = {
val implicits = spark.implicits
import implicits._
state.where("add IS NOT NULL").select($"add".as[AddFile])
}
这里调用了state方法，而它又调用了stateReconstruction方法，
private lazy val cachedState = cacheDS(stateReconstruction, s"Delta Table State #$version - $redactedpath") /** The current set of actions in this [[Snapshot]]. */ def state: Dataset[SingleAction] = cachedState.getDS
stateReconstruction方法在checkpoint的时用到了，@R_404_6329@也用到了，主要是重新构造文件状态，合并AddFile和RemoveFile：
private def stateReconstruction: Dataset[SingleAction] = {
...
loadActions.mapPartitions { actions =>
val hdpConf = hadoopConf.value.value
actions.flatMap {
_.unwrap match {
case add: AddFile => Some(add.copy(path = canonicalizePath(add.path, hdpConf)).wrap)
case rm: RemoveFile => Some(rm.copy(path = canonicalizePath(rm.path, hdpConf)).wrap)
case other if other == null => None
case other => Some(other.wrap)
}
}
}
...
.mapPartitions { iter =>
val state = new InMemoryLogReplay(time)
state.append(0, iter.map(_.unwrap))
state.checkpoint.map(_.wrap)
}
}
而关键在于InMemoryLogReplay的append方法和checkpoint方法，这里做到了文件状态的合并：
assert(currentVersion == -1 || version == currentVersion + 1,
s"Attempted to replay version $version, but state is at $currentVersion")
currentVersion = version
actions.foreach {
case a: SetTransaction =>
transactions(a.appId) = a
case a: Metadata =>
currentMetaData = a
case a: Protocol =>
currentProtocolVersion = a
case add: AddFile =>
activeFiles(add.pathAsUri) = add.copy(dataChange = false)
// Remove the tombstone to make sure we only output one `FileAction`.
tombstones.remove(add.pathAsUri)
case remove: RemoveFile =>
activeFiles.remove(remove.pathAsUri)
tombstones(remove.pathAsUri) = remove.copy(dataChange = false)
case ci: CommitInfo => // do nothing
case null => // Some crazy future feature. Ignore
}
}
重点就在case add: AddFile和 case remove: RemoveFile处理以及checkpoint方法，能够很好的合并文件状态。
再调用collect方法,返回DeltaScan，之后获取文件路径作为要处理的文件路径。
把TahoeLogFileIndex传入HadoopFsRelation得到最后的BaseRelation 返回
注意：spark读取delta格式整个流程和spark读取其他数据格式流程一致，主要区别在于读取数据之前，会把文件状态在内存中进行一次合并，这样只需要读取文件状态为Addfile的就行了
以上是“spark delta如何读数据”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注编程之家行业资讯频道！
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
