1. 背景

多維分析是大數(shù)據(jù)分析的一個典型場景，這種分析一般帶有過濾條件。對于此類查詢，尤其是在高基字段的過濾查詢，理論上只我們對原始數(shù)據(jù)做合理的布局，結(jié)合相關(guān)過濾條件，查詢引擎可以過濾掉大量不相關(guān)數(shù)據(jù)，只需讀取很少部分需要的數(shù)據(jù)。例如我們在入庫之前對相關(guān)字段做排序，這樣生成的每個文件相關(guān)字段的min－max值是不存在交叉的，查詢引擎下推過濾條件給數(shù)據(jù)源結(jié)合每個文件的min－max統(tǒng)計信息，即可過濾掉大量不相干數(shù)據(jù)。 上述技術(shù)即我們通常所說的data clustering 和 data skip。直接排序可以在單個字段上產(chǎn)生很好的效果，如果多字段直接排序那么效果會大大折扣的，Z-Order可以較好的解決多字段排序問題。

本文基于Apache Spark 以及 Apache Hudi 結(jié)合Z－order技術(shù)介紹如何更好的對原始數(shù)據(jù)做布局， 減少不必要的I/O，進而提升查詢速度。具體提案可參考Hudi RFC-28：Support Z-order curve

2. Z-Order介紹

Z-Order是一種可以將多維數(shù)據(jù)壓縮到一維的技術(shù)，在時空索引以及圖像方面使用較廣。Z曲線可以以一條無限長的一維曲線填充任意維度的空間，對于數(shù)據(jù)庫的一條數(shù)據(jù)來說，我們可以將其多個要排序的字段看作是數(shù)據(jù)的多個維度，z曲線可以通過一定的規(guī)則將多維數(shù)據(jù)映射到一維數(shù)據(jù)上，構(gòu)建z-value 進而可以基于該一維數(shù)據(jù)進行排序。z-value的映射規(guī)則保證了排序后那些在多維維度臨近的數(shù)據(jù)在一維曲線上仍然可以彼此臨近。

wiki定義：假設(shè)存在一個二維坐標對(x, y)，這些坐標對于于一個二維平面上，使用Z排序，我們可以將這些坐標對壓縮到一維。

當前在delta lake的商業(yè)版本實現(xiàn)了基于Z-Order的data Clustering技術(shù)，開源方面Spark/Hive/Presto 均未有對Z-Order的支持。

3. 具體實現(xiàn)

我們接下來分2部分介紹如何在Hudi中使用Z-Order：

• z-value的生成和排序
• 與Hudi結(jié)合

3.1 z-value的生成和排序

這部分是Z-Order策略的核心，這部分邏輯是公用的，同樣適用其他框架。

Z-Order的關(guān)鍵在于z-value的映射規(guī)則。wiki上給出了基于位交叉的技術(shù)，每個維度值的比特位交叉出現(xiàn)在最終的z-value里。例如假設(shè)我們想計算二維坐標（x=97, y=214）的z-value，我們可以按如下步驟進行

第一步：將每一維數(shù)據(jù)用bits表示

x value：01100001
y value：11010110

第二步：從y的最左側(cè)bit開始，我們將x和y按位做交叉，即可得到z 值，如下所示

z-value： 1011011000101001

對于多維數(shù)據(jù)，我們可以采用同樣的方法對每個維度的bit位做按位交叉形成 z-value，一旦我們生成z-values 我們即可用該值做排序，基于z值的排序自然形成z階曲線對多個參與生成z值的維度都有良好的聚合效果。

上述生成z-value的方法看起來非常好，但在實際生產(chǎn)環(huán)境上我們要使用位交叉技術(shù)產(chǎn)生z-value 還需解決如下問題：

• 上述介紹是基于多個unsigned int類型的遞增數(shù)據(jù)，通過位交叉生成z-value的。實際上的數(shù)據(jù)類型多種多樣，如何處理其他類型數(shù)據(jù)
• 不同類型的維度值轉(zhuǎn)成bit位表示，長度不一致如何處理
• 如何選擇數(shù)據(jù)類型合理的保存z-value，以及相應(yīng)的z值排序策略

針對上述問題，我們采用兩種策略生成z值。

3.1.1 基于映射策略的z值生成方法

第一個問題：對不同的數(shù)據(jù)類型采用不同的轉(zhuǎn)換策略

• 無符號類型整數(shù)： 直接轉(zhuǎn)換成bits位表示

• Int類型的數(shù)據(jù)： 直接轉(zhuǎn)成二進制表示會有問題，因為java里面負數(shù)的二進制表示最高位（符號位）為1，而正整數(shù)的二進制表示最高位為0（如下圖所示）， 直接轉(zhuǎn)換后會出現(xiàn)負數(shù)大于正數(shù)的現(xiàn)象。

對于這個問題，我們可以直接將二進制的最高位反轉(zhuǎn)，就可以保證轉(zhuǎn)換后的詞典順序和原值相同。如下圖

• Long類型的數(shù)據(jù)：轉(zhuǎn)換方式和Int類型一樣，轉(zhuǎn)成二進制形式并將最高位反轉(zhuǎn)
• Double、Float類型的數(shù)據(jù)： 轉(zhuǎn)成Long類型，之后轉(zhuǎn)成二進制形式并將最高位反轉(zhuǎn)
• Decimal/Date/TimeStamp類型數(shù)據(jù)：轉(zhuǎn)換成long類型，然后直接用二進制表示。
• UTF-8 String類型的數(shù)據(jù)：String類型的數(shù)據(jù) 直接用二進制表示即可保持原來的自然序， 但是字符串是不定長的無法直接用來做位交叉。 我們采用如下策略處理string類型大于8bytes的字符串截斷成8bytes， 不足8bytes的string 填充成8bytes。
• null值處理：
  • 數(shù)值類型的null直接變成該數(shù)值類型的最大值，之后按上述步驟轉(zhuǎn)換；
  • String類型null 直接變成空字符串之后再做轉(zhuǎn)換；

第二個問題：生成的二進制值統(tǒng)一按64位對齊即可

第三個問題：可以用Array[Byte]來保存z值（參考Amazon的DynamoDB 可以限制該數(shù)組的長度位1024）。對于 Array[Byte]類型的數(shù)據(jù)排序，hbase的rowkey 排序器可以直接拿來解決這個問題

基于映射策略的z值生成方法，方便快捷很容易理解，但是有一定缺陷：

參與生成z-value的字段理論上需要是從0開始的正整數(shù)，這樣才能生成很好的z曲線。 真實的數(shù)據(jù)集中 是不可能有這么完美的情況出現(xiàn)的， zorder的效果將會打折扣。比如x 字段取值(0, 1, 2)， y字段取值(100, 200, 300)， 用x, y生成的z-value只是完整z曲線的一部分，對其做z值排序的效果和直接用x排序的效果是一樣的； 再比如x的基數(shù)值遠遠低于y的基數(shù)值時采用上述策略排序效果基本和按y值排序是一樣的，真實效果還不如先按x排序再按y排序。

String類型的處理， 上述策略對string類型是取前8個字節(jié)的參與z值計算, 這將導(dǎo)致精度丟失。 當出現(xiàn)字符串都是相同字符串前綴的情況就無法處理了，比如"https://www.baidu.com" , "https://www.google.com" 這兩個字符串前8個字節(jié)完全一樣， 對這樣的數(shù)據(jù)截取前8個字節(jié)參與z值計算沒有任何意義。

上述策略出現(xiàn)缺陷的主要原因是數(shù)據(jù)的分布并不總是那么好導(dǎo)致。有一種簡單的方案可以解決上述問題： 對參與z值計算的所有維度值做全局Rank，用Rank值代替其原始值參與到z值計算中，由于Rank值一定是從0開始的正整數(shù)，完全符合z值構(gòu)建條件，較好的解決上述問題。 在實驗中我們發(fā)現(xiàn)這種用Rank值的方法確實很有效，但是z值生成效率極低，計算引擎做全局Rank的代價是非常高的，基于Rank的方法效率瓶頸在于要做全局Rank計算，那么我們可不可以對原始數(shù)據(jù)做采樣減少數(shù)據(jù)量，用采樣后的數(shù)據(jù)計算z值呢，答案是肯定的。

/** Generates z-value*/
val newRDD = df.rdd.map { row =>
  val values = zFields.map { case (index, field) =>
    field.dataType match {
      case LongType =>
        ZOrderingUtil.longTo8Byte(row.getLong(index))
      case DoubleType =>
        ZOrderingUtil.doubleTo8Byte(row.getDouble(index))
      case IntegerType =>
        ZOrderingUtil.intTo8Byte(row.getInt(index))
      case FloatType =>
        ZOrderingUtil.doubleTo8Byte(row.getFloat(index).toDouble)
      case StringType =>
        ZOrderingUtil.utf8To8Byte(row.getString(index))
      case DateType =>
        ZOrderingUtil.longTo8Byte(row.getDate(index).getTime)
      case TimestampType =>
        ZOrderingUtil.longTo8Byte(row.getTimestamp(index).getTime)
      case ByteType =>
        ZOrderingUtil.byteTo8Byte(row.getByte(index))
      case ShortType =>
        ZOrderingUtil.intTo8Byte(row.getShort(index).toInt)
      case d: DecimalType =>
        ZOrderingUtil.longTo8Byte(row.getDecimal(index).longValue())
      case _ =>
        null
    }
  }.filter(v => v != null).toArray
  val zValues = ZOrderingUtil.interleaveMulti8Byte(values)
  Row.fromSeq(row.toSeq ++ Seq(zValues))
}.sortBy(x => ZorderingBinarySort(x.getAs[Array[Byte]](fieldNum)))

3.1.2 基于RangeBounds的z-value生成策略

在介紹基于RangeBounds的z-value生成策略之前先看看Spark的排序過程，Spark排序大致分為2步

• 對輸入數(shù)據(jù)的key做sampling來估計key的分布，按指定的分區(qū)數(shù)切分成range并排序。計算出來的rangeBounds是一個長度為numPartition - 1 的數(shù)組，該數(shù)組里面每個元素表示一個分區(qū)內(nèi)key值的上界/下界。
• shuffle write 過程中，每個輸入的key應(yīng)該分到哪個分區(qū)內(nèi)，由第一步計算出來的rangeBounds來確定。每個分區(qū)內(nèi)的數(shù)據(jù)雖然沒有排序，但是注意rangeBounds是有序的因此分區(qū)之間宏觀上看是有序的，故只需對每個分區(qū)內(nèi)數(shù)據(jù)做好排序即可保證數(shù)據(jù)全局有序。

參考Spark的排序過程，我們可以這樣做

• 對每個參與Z-Order的字段篩選規(guī)定個數(shù)（類比分區(qū)數(shù)）的Range并對進行排序，并計算出每個字段的RangeBounds；
• 實際映射過程中每個字段映射為該數(shù)據(jù)所在rangeBounds的中的下標，然后參與z-value的計算?？梢钥闯鲇捎趨^(qū)間下標是從0開始遞增的正整數(shù)，完全滿足z值生成條件；并且String類型的字段映射問題也被一并解決了。基于RangeBounds的z值生成方法，很好的解決了第一種方法所面臨的缺陷。由于多了一步采樣生成RangeBounds的過程，其效率顯然不如第一種方案，我們實現(xiàn)了上述兩種z值生成方法以供選擇。

/** Generates z-value */
val indexRdd = internalRdd.mapPartitionsInternal { iter =>
  val bounds = boundBroadCast.value
  val origin_Projections = sortingExpressions.map { se =>
    UnsafeProjection.create(Seq(se), outputAttributes)
  }
  iter.map { unsafeRow =>
    val interleaveValues = origin_Projections.zip(origin_lazyGeneratedOrderings).zipWithIndex.map { case ((rowProject, lazyOrdering), index) =>
      val row = rowProject(unsafeRow)
      val decisionBound = new DecisionBound(sampleRdd, lazyOrdering)
      if (row.isNullAt(0)) {
        bounds(index).length + 1
      } else {
        decisionBound.getBound(row, bounds(index).asInstanceOf[Array[InternalRow]])
      }
    }.toArray.map(ZOrderingUtil.toBytes(_))
    val zValues = ZOrderingUtil.interleaveMulti4Byte(interleaveValues)
    val mutablePair = new MutablePair[InternalRow, Array[Byte]]()
    mutablePair.update(unsafeRow, zValues)
  }
}.sortBy(x => ZorderingBinarySort(x._2), numPartitions = fileNum).map(_._1)

3.2 與Hudi結(jié)合

與Hudi的結(jié)合大致分為兩部分

3.2.1 表數(shù)據(jù)的Z排序重組

這塊相對比較簡單，借助Hudi內(nèi)部的Clustering機制結(jié)合上述z值的生成排序策略我們可以直接完成Hudi表數(shù)據(jù)的數(shù)據(jù)重組，這里不再詳細介紹。

3.2.2 收集保存統(tǒng)計信息

這塊其實RFC27已經(jīng)在做了，感覺有點重復(fù)工作我們簡單介紹下我們的實現(xiàn)，數(shù)據(jù)完成z重組后，我們需要對重組后的每個文件都收集參與z值計算的各個字段的min/max/nullCount 的統(tǒng)計信息。對于統(tǒng)計信息收集，可以通過讀取Parquet文件或者通過SparkSQL收集

• 讀取Parquet文件收集統(tǒng)計信息

/** collect statistic info*/
val sc = df.sparkSession.sparkContext
val serializableConfiguration = new SerializableConfiguration(conf)
val numParallelism = inputFiles.size/3
val previousJobDescription = sc.getLocalProperty(SparkContext.SPARK_JOB_DESCRIPTION)
try {
  val description = s"Listing parquet column statistics"
  sc.setJobDescription(description)
  sc.parallelize(inputFiles, numParallelism).mapPartitions { paths =>
    val hadoopConf = serializableConfiguration.value
    paths.map(new Path(_)).flatMap { filePath =>
      val blocks = ParquetFileReader.readFooter(hadoopConf, filePath).getBlocks().asScala
      blocks.flatMap(b => b.getColumns().asScala.
        map(col => (col.getPath().toDotString(),
          FileStats(col.getStatistics().minAsString(), col.getStatistics().maxAsString(), col.getStatistics.getNumNulls.toInt))))
        .groupBy(x => x._1).mapValues(v => v.map(vv => vv._2)).
        mapValues(value => FileStats(value.map(_.minVal).min, value.map(_.maxVal).max, value.map(_.num_nulls).max)).toSeq.
        map(x => ColumnFileStats(filePath.getName(), x._1, x._2.minVal, x._2.maxVal, x._2.num_nulls))
    }.filter(p => cols.contains(p.colName))
  }.collect()
} finally {
  sc.setJobDescription(previousJobDescription)
}

• 通過SparkSQL方式收集統(tǒng)計信息

/** collect statistic info*/
val inputFiles = df.inputFiles
val conf = df.sparkSession.sparkContext.hadoopConfiguration
val values = cols.flatMap(c => Seq( min(col(c)).as(c + "_minValue"), max(col(c)).as(c + "_maxValue"), count(c).as(c + "_noNullCount")))
val valueCounts = count("*").as("totalNum")
val projectValues = Seq(col("file")) ++ cols.flatMap(c =>
  Seq(col(c + "_minValue"), col(c + "_maxValue"), expr(s"totalNum - ${c + "_noNullCount"}").as(c + "_num_nulls")))
val result = df.select(input_file_name() as "file", col("*"))
  .groupBy($"file")
  .agg(valueCounts,  values: _*).select(projectValues:_*)
result

之后將這些信息保存在Hudi表里面的hoodie目錄下的index目錄下，然后供Spark查詢使用。

3.2.3 應(yīng)用到Spark查詢

為將統(tǒng)計信息應(yīng)用Spark查詢，需修改HudiIndex的文件過濾邏輯，將DataFilter轉(zhuǎn)成對Index表的過濾，選出候選要讀取的文件，返回給查詢引擎，具體步驟如下。

• 將索引表加載到 IndexDataFrame
• 使用原始查詢過濾器為 IndexDataFrame 構(gòu)建數(shù)據(jù)過濾器
• 查詢 IndexDataFrame 選擇候選文件
• 使用這些候選文件來重建 HudiMemoryIndex

通過min/max值和null計數(shù)信息為 IndexDataFrame 構(gòu)建數(shù)據(jù)過濾器，由于z排序后參與z值計算的各個字段在每個文件里面的min/max值很大概率不交叉，因此對Index表的過濾可以過濾掉大量的文件。

/** convert filter */
def createZindexFilter(condition: Expression): Expression = {
  val minValue = (colName: Seq[String]) =>
    col(UnresolvedAttribute(colName) + "_minValue").expr
  val maxValue = (colName: Seq[String]) =>
    col(UnresolvedAttribute(colName) + "_maxValue").expr
  val num_nulls = (colName: Seq[String]) =>
    col(UnresolvedAttribute(colName) + "_num_nulls").expr
  condition match {
    case EqualTo(attribute: AttributeReference, value: Literal) =>
      val colName = HudiMergeIntoUtils.getTargetColNameParts(attribute)
      And(LessThanOrEqual(minValue(colName), value), GreaterThanOrEqual(maxValue(colName), value))
    case EqualTo(value: Literal, attribute: AttributeReference) =>
      val colName = HudiMergeIntoUtils.getTargetColNameParts(attribute)
      And(LessThanOrEqual(minValue(colName), value), GreaterThanOrEqual(maxValue(colName), value))
    case equalNullSafe @ EqualNullSafe(_: AttributeReference, _ @ Literal(null, _)) =>
      val colName = HudiMergeIntoUtils.getTargetColNameParts(equalNullSafe.left)
      EqualTo(num_nulls(colName), equalNullSafe.right)
.......

4. 測試結(jié)果

我們采用databrick的測試樣例https://help.aliyun.com/document_detail/168137.html?spm=a2c4g.11186623.6.563.53c258ccmqvYfy 進行了測試

測試數(shù)據(jù)量和資源使用大小和databrick保持一致。唯一區(qū)別是我們只生成了10000個文件，原文是100w個文件。 測試結(jié)果表明zorder加速比還說很可觀的，另外Z-Order的效果隨著文件數(shù)的增加會越來越好，我們后續(xù)也會在100w文件級別測試。

以上就是Z-Order加速Hudi大規(guī)模數(shù)據(jù)集方案分析的詳細內(nèi)容，更多關(guān)于Z-Order加速Hudi大規(guī)模數(shù)據(jù)集的資料請關(guān)注腳本之家其它相關(guān)文章！