歡迎關(guān)注公眾號(hào)“Tim在路上”
通常為提高數(shù)據(jù)處理的效率，計(jì)算引擎要實(shí)現(xiàn)謂詞的下推，而存儲(chǔ)引擎可以根據(jù)下推的過濾條件盡可能的跳過無(wú)關(guān)數(shù)據(jù)或文件。不管是Hudi、Iceberg還是Delta都實(shí)現(xiàn)了基于min-max索引的Data-skiping技術(shù)。它指的是在元數(shù)據(jù)中都記錄這數(shù)據(jù)文件中的每一列的最小值和最大值，通過查詢中列上的謂詞來決定當(dāng)前的數(shù)據(jù)文件是否可能包含滿足謂詞的任何records，是否可以跳過讀取當(dāng)前數(shù)據(jù)文件。
但是當(dāng)當(dāng)數(shù)據(jù)均勻分布在所有文件中時(shí)，那么每個(gè)文件列的upper_bounds和lower_bounds的range會(huì)很大，那么這時(shí)數(shù)據(jù)跳過的能力就會(huì)失效。其次為了避免分區(qū)字段列與其他查詢過濾列存在clustering或相關(guān)性，一般是建議在查詢前進(jìn)行sort排序。
但是傳統(tǒng)的線性排序排序，其跳過效率僅在第一列中很高，但在隨后的列中其效果迅速下降。因此，如果有兩個(gè)或更多列同樣可能出現(xiàn)在高度選擇性的謂詞中，則數(shù)據(jù)跳過將無(wú)法為這個(gè)整體帶來更好的性能。
從上面圖片中的例子可以看出, 對(duì)于按字典順序排列的 3 元組整數(shù)，只有第一列能夠通過排序?qū)?shù)據(jù)聚集起來變成連續(xù)可篩選的數(shù)據(jù)，但是，如果在第三列中找到值為“4”的數(shù)據(jù)，就會(huì)發(fā)現(xiàn)它現(xiàn)在分散在各處，根本沒有本地化，那數(shù)據(jù)的跳過性能可想而知。
Z-order 又名Z階曲線有個(gè)非常重要的應(yīng)用特性，就是降維。它可以將多維空間問題降維到低維或者一維空間問題。將多列轉(zhuǎn)換為一個(gè)Z-index列，按照其進(jìn)行排序，根據(jù)Z-Order值相近的數(shù)據(jù)會(huì)分布到同一個(gè)文件中的特性，從各個(gè)維度的值分布來說，從數(shù)據(jù)整體來看也會(huì)呈現(xiàn)近似單調(diào)的分布。但是，文件的upper_bounds和lower_bounds的重合度會(huì)有效降低，dataskipping技術(shù)又可以重新生效。 Z-order的核心是提高做File Skip 而非 Row Skip， 這樣更能減少不必要的IO。除了z-order之外常見的還有希爾伯特曲線（Hilbert curve）。
Z-order 簡(jiǎn)要說明
映射多維數(shù)據(jù)到一維，并按照這個(gè)維度進(jìn)行排序
Z-Order的關(guān)鍵在于z-value的映射規(guī)則?；谖唤徊娴募夹g(shù)，每個(gè)維度值的比特位交叉出現(xiàn)在最終的z-value里。例如假設(shè)我們想計(jì)算二維 坐標(biāo)（x=97, y=214）的z-value，我們可以按如下步驟進(jìn)行
第一步：將每一維數(shù)據(jù)用bits表示
x value：01100001 97 98
y value：11010110 104 105
第二步：從y的最左側(cè)bit開始，我們將x和y按位做交叉，即可得到z 值，如下所示
z-value： 1011011000101001 46633
對(duì)于多維數(shù)據(jù)，我們可以采用同樣的方法對(duì)每個(gè)維度的bit位做按位交叉形成 z-value，一旦我們生成z-values 我們即可用該值做排序，基于z值的排序自然形成z階曲線對(duì)多個(gè)參與生成z值的維度都有良好的聚合效果。
zorder的性能效果如何呢？下面我們舉個(gè)例子：
[圖片上傳失敗...(image-eda57c-1657366659242)]
在上面的圖片中，每個(gè)數(shù)據(jù)框代表一個(gè)文件，每個(gè)文件均勻存放4個(gè)數(shù)據(jù)，左邊是線性排序后的數(shù)據(jù)分布，右邊是Zorder排序。從中可以看出在查詢x = 2 or y = 2的條件時(shí)，線性排序需要掃描9個(gè)文件，zorder排序只需要掃描7個(gè)文件。
Delta的Z-order 的幾個(gè)細(xì)節(jié)
可以說實(shí)現(xiàn)Z-order并不難，但實(shí)現(xiàn)高效的Z-order還是比較復(fù)雜的。要實(shí)現(xiàn)Z-order, 首先就要考慮如何將多列查詢謂詞值轉(zhuǎn)換為z-value。
從上面的介紹可以看出要生成z-value，目前最直觀的辦法是將多維數(shù)據(jù)轉(zhuǎn)換為二進(jìn)制然后再進(jìn)行按位交叉生成z-value。如果直接將不同類型的數(shù)據(jù)轉(zhuǎn)換為二進(jìn)制，那么會(huì)存在幾個(gè)問題：
- 如何保證不同類型的維度值(String, Long, Double ...)轉(zhuǎn)成bit位時(shí)長(zhǎng)度一致？這里可能需要對(duì)位數(shù)不夠的進(jìn)行左填充補(bǔ)0，另外對(duì)于String這類比較長(zhǎng)的可能需要進(jìn)行截取。
- 不同數(shù)據(jù)類型的null值如何處理？z-value的交叉生成不允許存在null值，這里可以選取min-max值作為null的填充。
從上面可以看出如果直接將多列值轉(zhuǎn)換為二進(jìn)制，不僅需要為每列值分配新的字節(jié)緩沖區(qū)，還需要對(duì)不同的數(shù)據(jù)類型進(jìn)行不同的額外操作，同時(shí)由于String截取的存在可能造成數(shù)據(jù)不精準(zhǔn)的存在, 而String類型又是比較常用的類型。
為了解決上述問題，一般采用對(duì)查詢列進(jìn)行排序，將每行數(shù)據(jù)映射為順序id, 類似于row_number()或dense_rank()或rank()的窗口函數(shù)。
然而這種情況下對(duì)查詢列進(jìn)行依次排序，可見性能上肯定影響很大。
那么Delta是如何實(shí)現(xiàn)的？又是如何解決上述問題的？
Delta采取了降低精度的辦法，將連續(xù)的值視為一個(gè)單位，將任意的查詢列轉(zhuǎn)換為range_parition_id()。這里的分區(qū)數(shù)可以用OPTIMIZE_ZORDERBY_NUM_RANGE_IDS表示。
那么如何實(shí)現(xiàn)呢？實(shí)現(xiàn)其實(shí)非常簡(jiǎn)單，獲取range_parition_id的實(shí)際上只需要重用了Spark中的RangePartition操作，并在其基礎(chǔ)上實(shí)現(xiàn)了range_partition_id(col, N) -> int的表達(dá)式。通過這個(gè)表達(dá)式就實(shí)現(xiàn)了將查詢類轉(zhuǎn)換為二進(jìn)制的過程，這個(gè)過程避免了額外操作以及多次排序。這樣的實(shí)現(xiàn)利用RangePartition對(duì)鍵進(jìn)行采樣計(jì)算分區(qū)邊界的實(shí)現(xiàn)。
將多個(gè)查詢列轉(zhuǎn)換為二級(jí)制后，然后通過調(diào)用interleace_bits(...)交叉的方法，就生成了Z-value。
那么如何進(jìn)行排序?qū)懗鰧?shí)現(xiàn)呢？采用那種方式呢？
如何直接將數(shù)據(jù)按照Z(yǔ)-value進(jìn)行全局排序，會(huì)存在兩個(gè)問題：
- 對(duì)整個(gè)數(shù)據(jù)排序是非常低效的。
- Z-order曲線中存在“接縫”，其中需要線性遍歷在繼續(xù)其路徑之前跳轉(zhuǎn)到不同的區(qū)域， 這很不利于小范圍內(nèi)的查詢。
那么Delta實(shí)現(xiàn)主要是將其按照z-value進(jìn)行range分區(qū)，實(shí)際上就是調(diào)用了Spark的repartitionByRange的表達(dá)式。
如何處理數(shù)據(jù)傾斜呢？
如果要聚類的列整體上是傾斜的，那么即使轉(zhuǎn)換為z-value也會(huì)是傾斜的，這時(shí)候如果對(duì)其進(jìn)行排序?qū)懗隹赡軙?huì)比較耗時(shí)。這里的解決辦法其實(shí)很簡(jiǎn)單就是在z-value字節(jié)數(shù)組的結(jié)尾追加隨機(jī)字節(jié)，然后再對(duì)其進(jìn)行分區(qū)范圍內(nèi)排序。
Delta的Z-order源碼分析
下面我們從用戶調(diào)用的角度來分析下源碼：
OPTIMIZE delta.table WHERE day = 25 AND city = ‘New York’ ZORDER BY (col1, col2)
- SQL解析為Optimizer命令，進(jìn)行執(zhí)行前校驗(yàn)
當(dāng)用戶執(zhí)行一條上面的sql后，首先會(huì)經(jīng)歷sql解析階段。Spark使用的是開源組件antlr4將輸入SQL解析為AST樹。它的解析語(yǔ)法在DeltaSQLBase.g4文件中。
| OPTIMIZE (path=STRING | table=qualifiedName)
(WHERE partitionPredicate = predicateToken)?
(zorderSpec)? #optimizeTable
zorderSpec
: ZORDER BY LEFT_PAREN interleave+=qualifiedName (COMMA interleave+=qualifiedName)* RIGHT_PAREN
| ZORDER BY interleave+=qualifiedName (COMMA interleave+=qualifiedName)*
;
從上面的源碼可以看出，OPTIMIZE sql不僅支持表名還支持直接指定的優(yōu)化文件目錄。但這里要注意的是在優(yōu)化數(shù)據(jù)布局的時(shí)候，where條件的過濾列必須分區(qū)分區(qū)列的子集。即查詢列day和city必須是分區(qū)列。
解析g4文件一般是在DeltaSqlParser類中，通過visitZorderSpec方法就可以拿到用戶輸入的zorder列。而要優(yōu)化的目錄名和表名以及過濾的條件是通過visitOptimizeTable方法獲取的。
override def visitOptimizeTable(ctx: OptimizeTableContext): AnyRef = withOrigin(ctx) {
if (ctx.path == null && ctx.table == null) {
throw new ParseException("OPTIMIZE command requires a file path or table name.", ctx)
}
// z-order的列的seq
val interleaveBy = Option(ctx.zorderSpec).map(visitZorderSpec).getOrElse(Seq.empty)
OptimizeTableCommand(
Option(ctx.path).map(string),
Option(ctx.table).map(visitTableIdentifier),
Option(ctx.partitionPredicate).map(extractRawText(_)))(interleaveBy)
}
這里通過訪問者的設(shè)計(jì)模式，獲取OptimizeTableContext命令中的數(shù)據(jù)。從上面可以看出先從visitZorderSpec獲取z-order的列的數(shù)組，然后將其封裝到OptimizeTableCommand類中。OptimizeTableCommand屬于command表達(dá)式，它在執(zhí)行時(shí)會(huì)執(zhí)行其run方法。
override def run(sparkSession: SparkSession): Seq[Row] = {
val deltaLog = getDeltaLog(sparkSession, path, tableId, "OPTIMIZE")
// [1] 從metadata中獲取表的分區(qū)列
val partitionColumns = deltaLog.snapshot.metadata.partitionColumns
// [2] 解析查詢謂詞
val partitionPredicates = partitionPredicate.map(predicate => {
val predicates = parsePredicates(sparkSession, predicate)
verifyPartitionPredicates(
sparkSession,
partitionColumns,
predicates)
predicates
}).getOrElse(Seq.empty)
// [3] 校驗(yàn)zorder列
validateZorderByColumns(sparkSession, deltaLog, zOrderBy)
val zOrderByColumns = zOrderBy.map(_.name).toSeq
// [4] 調(diào)用optimize命令
new OptimizeExecutor(sparkSession, deltaLog, partitionPredicates, zOrderByColumns)
.optimize()
}
從這里可以看出在Optimize執(zhí)行前會(huì)先進(jìn)行校驗(yàn)，首先zorder應(yīng)該是對(duì)應(yīng)的分區(qū)目錄內(nèi)的數(shù)據(jù)進(jìn)行zorder, 所以zorder列不應(yīng)該包含分區(qū)列。其次zorder列，必須是在元數(shù)據(jù)中完成了min-max統(tǒng)計(jì)的列，即可以通過其進(jìn)行數(shù)據(jù)跳過。最后在調(diào)用OptimizeExecutor的optimize方法。下面我們到optimize方法中展開看看：
- 篩選候選文件， 并對(duì)文件進(jìn)行分區(qū)壓縮
def optimize(): Seq[Row] = {
val txn = deltaLog.startTransaction()
// [1] 按照where條件進(jìn)行篩選出候選文件
val candidateFiles = txn.filterFiles(partitionPredicate)
val partitionSchema = txn.metadata.partitionSchema
// [2] 注意這里如果isMultiDimClustering多維聚集則不過濾文件的大小直接選擇所有的文件
// select all files in case of multi-dimensional clustering
val filesToProcess = candidateFiles.filter(_.size < minFileSize || isMultiDimClustering)
val partitionsToCompact = filesToProcess.groupBy(_.partitionValues).toSeq
val jobs = groupFilesIntoBins(partitionsToCompact, maxFileSize)
這里主要進(jìn)行候選文件的篩選，同時(shí)在優(yōu)化前進(jìn)行文件分組。這里需要注意的是如果是多維聚集則不過濾文件的大小直接選擇所有的文件。這里的文件分組算法采用的壓縮采用的binpack算法，保證每個(gè)分組的文件size和均勻。
val parallelJobCollection = new ParVector(jobs.toVector)
// Create a task pool to parallelize the submission of optimization jobs to Spark.
val threadPool = ThreadUtils.newForkJoinPool(
"OptimizeJob",
sparkSession.sessionState.conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_MAX_THREADS))
val updates = try {
val forkJoinPoolTaskSupport = new ForkJoinTaskSupport(threadPool)
parallelJobCollection.tasksupport = forkJoinPoolTaskSupport
// 并行的執(zhí)行文件的合并、壓縮與Zorder優(yōu)化
parallelJobCollection.flatMap(partitionBinGroup =>
runOptimizeBinJob(txn, partitionBinGroup._1, partitionBinGroup._2, maxFileSize)).seq
} finally {
threadPool.shutdownNow()
}
然后并發(fā)的執(zhí)行文件的合并、壓縮與Zorder優(yōu)化，合并與壓縮當(dāng)然和zorder沒有什么關(guān)系，這是Optimizer原有功能的重用，可以優(yōu)化zorder排序后的性能。下面我們進(jìn)入runOptimizeBinJob方法中，主要看看Zorder優(yōu)化的實(shí)現(xiàn)。
- 根據(jù)多維列值生成Z-value
// [1] 讀取分組的文件
val input = txn.deltaLog.createDataFrame(txn.snapshot, bin, actionTypeOpt = Some("Optimize"))
val repartitionDF = if (isMultiDimClustering) {
val totalSize = bin.map(_.size).sum
// 分區(qū)數(shù)為分組文件的size大小除以其中最大的文件size
val approxNumFiles = Math.max(1, totalSize / maxFileSize).toInt
// [2] 調(diào)用 MultiDimClustering.cluster
MultiDimClustering.cluster(
input,
approxNumFiles,
zOrderByColumns)
} else {
input.coalesce(numPartitions = 1)
}
合并后的文件數(shù)為分組文件的size大小除以其中最大的文件size。這里就是讀取分組文件，然后調(diào)用cluster方法。
override def cluster(
df: DataFrame,
colNames: Seq[String],
approxNumPartitions: Int): DataFrame = {
val conf = df.sparkSession.sessionState.conf
// 用于控制rangeId，越大，精度越好，但以犧牲性能為代價(jià)
val numRanges = conf.getConf(DeltaSQLConf.MDC_NUM_RANGE_IDS)
// 是否追加噪音，為了避免數(shù)據(jù)傾斜的追加的噪音后綴
val addNoise = conf.getConf(DeltaSQLConf.MDC_ADD_NOISE)
val cols = colNames.map(df(_))
// 執(zhí)行cluster表達(dá)式，生成z-value
val mdcCol = getClusteringExpression(cols, numRanges)
val repartitionKeyColName = s"${UUID.randomUUID().toString}-rpKey1"
...
}
在cluster中，獲取numRanges和addNoise配置，然后再調(diào)用getClusteringExpression來獲取z-value列。
object ZOrderClustering extends SpaceFillingCurveClustering {
override protected[skipping] def getClusteringExpression(
cols: Seq[Column], numRanges: Int): Column = {
assert(cols.size >= 1, "Cannot do Z-Order clustering by zero columns!")
// [1] 調(diào)用range_partition_id表達(dá)式生成rangeIdCols
val rangeIdCols = cols.map(range_partition_id(_, numRanges))
// [2] 執(zhí)行interleave_bits，并轉(zhuǎn)換為String
interleave_bits(rangeIdCols: _*).cast(StringType)
}
}
[1] 調(diào)用range_partition_id表達(dá)式生成rangeIdCols
[2] 執(zhí)行interleave_bits，并轉(zhuǎn)換為String，這就是最終生成的z-value
range_partition_id函數(shù)就是range_partition_id(col, N) -> int的實(shí)現(xiàn)過程，通過上面的分區(qū)其實(shí)其是重用了Spark的RangePartition下面我們展開看看，這里是如何調(diào)用的。
def range_partition_id(col: Column, numPartitions: Int): Column = withExpr {
RangePartitionId(col.expr, numPartitions)
}
range_partition_id的實(shí)現(xiàn)非常簡(jiǎn)單，只是簡(jiǎn)單的將其封裝為RangePartitionId類并返回，RangePartitionId類是一個(gè)空的表達(dá)式操作。那么它是如果調(diào)用RangePartition的呢？
其實(shí)這個(gè)涉及到了SparkSQL的執(zhí)行優(yōu)化過程，SQL在執(zhí)行前，通常需要先進(jìn)行RBO優(yōu)化，CBO等優(yōu)化過程，這些優(yōu)化的實(shí)現(xiàn)通常以Rule的形式進(jìn)行注冊(cè)封裝，優(yōu)化后才轉(zhuǎn)換為RDD再執(zhí)行Spark任務(wù)。
extensions.injectOptimizerRule { session =>
new RangePartitionIdRewrite(session)
}
上面的代碼就是在優(yōu)化器中注入了一個(gè)RangePartitionIdRewrite規(guī)則，用于重寫 range_partition_id的占位符來調(diào)用RangePartitioner。
case RangePartitionId(expr, n) =>
val aliasedExpr = Alias(expr, "__RPI_child_col__")()
val exprAttr = aliasedExpr.toAttribute
// [1] 對(duì)于查詢列過濾null的行
val planForSampling = Filter(IsNotNull(exprAttr), Project(Seq(aliasedExpr), node.child))
...
withCallSite(session.sparkContext, desc) {
SQLExecution.withNewExecutionId(qeForSampling) {
withJobGroup(session.sparkContext, jobGroupId, desc) {
// [2] 創(chuàng)建一個(gè)pair(InternalRow, null), 用于存儲(chǔ)查詢列對(duì)應(yīng)的rangeid
val rddForSampling = qeForSampling.toRdd.mapPartitionsInternal { iter =>
val mutablePair = new MutablePair[InternalRow, Null]()
iter.map(row => mutablePair.update(row.copy(), null))
}
// [3] 創(chuàng)建RangePartitioner，傳入排序的sortOrder
val sortOrder = SortOrder(exprAttr, Ascending)
implicit val ordering = new LazilyGeneratedOrdering(Seq(sortOrder), Seq(exprAttr))
val partitioner = new RangePartitioner(n, rddForSampling, true, sampleSizeHint)
// [4] 調(diào)用PartitionerExpr，執(zhí)行寫入rangeid
PartitionerExpr(expr, partitioner)
override def eval(input: InternalRow): Any = {
val value: Any = child.eval(input)
row.update(0, value)
partitioner.getPartition(row)
}
從上面的代碼可以看出這里主要做了幾件事：
[1] 對(duì)于查詢列過濾null的行
[2] 創(chuàng)建一個(gè)pair(InternalRow, null), 用于存儲(chǔ)查詢列對(duì)應(yīng)的rangeid
[3] 創(chuàng)建RangePartitioner，傳入排序的sortOrder
[4] 調(diào)用PartitionerExpr，執(zhí)行寫入rangeid
- 根據(jù)z-value進(jìn)行range重分區(qū)
下面我們?cè)倩氐絚luster方法中，看看剩余的代碼：
var repartitionedDf = if (addNoise) {
val randByteColName = s"${UUID.randomUUID().toString}-rpKey2"
val randByteCol = (rand() * 255 - 128).cast(ByteType)
df.withColumn(repartitionKeyColName, mdcCol).withColumn(randByteColName, randByteCol)
.repartitionByRange(approxNumPartitions, col(repartitionKeyColName), col(randByteColName))
.drop(randByteColName)
} else {
df.withColumn(repartitionKeyColName, mdcCol)
.repartitionByRange(approxNumPartitions, col(repartitionKeyColName))
}
repartitionedDf.drop(repartitionKeyColName)
這里的代碼就非常直觀了，其實(shí)際上就是調(diào)用repartitionByRange表達(dá)式，并最終將z-value傳入，最終再將拼接的排序分區(qū)列刪除。最后再調(diào)用txn.writeFiles(repartitionDF)進(jìn)行執(zhí)行。
- 更新統(tǒng)計(jì)信息
if (isMultiDimClustering) {
val inputFileStats =
ZOrderFileStats(removedFiles.size, removedFiles.map(_.size.getOrElse(0L)).sum)
optimizeStats.zOrderStats = Some(ZOrderStats(
strategyName = "all", // means process all files in a partition
inputCubeFiles = ZOrderFileStats(0, 0),
inputOtherFiles = inputFileStats,
inputNumCubes = 0,
mergedFiles = inputFileStats,
// There will one z-cube for each partition
numOutputCubes = optimizeStats.numPartitionsOptimized))
}
最后在更新進(jìn)行合并、壓縮和zorder排序后文件的統(tǒng)計(jì)信息。
下面我們來總結(jié)下整個(gè)過程，并對(duì)比下和Iceberg、Hudi的實(shí)現(xiàn)區(qū)別：
- 需要篩選出待優(yōu)化的文件。OPTIMIZE語(yǔ)句的where條件只支持使用分區(qū)列，也就是支持對(duì)表的某些分區(qū)進(jìn)行OPTIMIZE。
- 根據(jù)多維列值計(jì)算出Z地址。這里將不同類型查詢列轉(zhuǎn)換為粗放的rangeId, 然后將查詢各列的rangId轉(zhuǎn)換為二進(jìn)制進(jìn)行交叉組合生成z-value。但是這里的rangeId需要通過專家經(jīng)驗(yàn)的配置，其次其解決數(shù)據(jù)傾斜時(shí)在z-value數(shù)組中隨機(jī)追加噪音字節(jié)。
- 根據(jù)z-value進(jìn)行range重分區(qū)。數(shù)據(jù)會(huì)shuffle到多個(gè)partition中。這一步等價(jià)于repartitionByRange(z-value)。
- 將重分區(qū)的partition使用Copy on Write寫回到存儲(chǔ)系統(tǒng)中，然后更新統(tǒng)計(jì)信息。
從這里可以看出Delta實(shí)現(xiàn)的z-order和Hudi、Iceberg的實(shí)現(xiàn)從本質(zhì)上來說都是文件間的排序采用zorder， 文件內(nèi)的排序任然采用線性排序的思想。這樣可以避免在小范圍查詢中（查詢正好落入單個(gè)文件內(nèi)）使用線性排序會(huì)有更好的性能。但是不同的是生成z-value的方式上的不同，Delta生成z-value的方式是采用映射為rangeid的辦法，并未采用直接轉(zhuǎn)換為二進(jìn)制的辦法。這種方式避免了額外操作以及多次排序，但需要更多的專家經(jīng)驗(yàn)。另外Delta的Zorder操作需要用戶手動(dòng)的執(zhí)行優(yōu)化。
下面我們留下幾個(gè)問題，可以思考下：
- Z-order排序的列一般選擇那些列進(jìn)行排序優(yōu)化，是否排序的列越多越好？
- Z-order排序后，是否對(duì)所有的查詢sql有提速的效果，那些場(chǎng)景會(huì)不會(huì)變的更慢？
