Z-Order加速Hudi大規模數據集的方法

本篇內容介紹了“Z-Order加速Hudi大規模數據集的方法”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

1. 背景

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

2. Z-Order介紹

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

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

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

3. 具體實現

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

• z-value的生成和排序

• 與Hudi結合

3.1 z-value的生成和排序

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

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

第一步：將每一維數據用bits表示

x value：01100001
y value：11010110

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

z-value： 1011011000101001

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

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

• 上述介紹是基于多個unsigned int類型的遞增數據，通過位交叉生成z-value的。實際上的數據類型多種多樣，如何處理其他類型數據

• 不同類型的維度值轉成bit位表示，長度不一致如何處理

• 如何選擇數據類型合理的保存z-value，以及相應的z值排序策略

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

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

第一個問題：對不同的數據類型采用不同的轉換策略

• 無符號類型整數： 直接轉換成bits位表示

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

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

• Long類型的數據：轉換方式和Int類型一樣，轉成二進制形式并將最高位反轉

• Double、Float類型的數據： 轉成Long類型，之后轉成二進制形式并將最高位反轉

• Decimal/Date/TimeStamp類型數據：轉換成long類型，然后直接用二進制表示。

• UTF-8 String類型的數據：String類型的數據 直接用二進制表示即可保持原來的自然序， 但是字符串是不定長的無法直接用來做位交叉。 我們采用如下策略處理string類型大于8bytes的字符串截斷成8bytes， 不足8bytes的string 填充成8bytes。

• null值處理：

• 數值類型的null直接變成該數值類型的最大值，之后按上述步驟轉換；

• String類型null 直接變成空字符串之后再做轉換；

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

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

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

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

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

上述策略出現缺陷的主要原因是數據的分布并不總是那么好導致。有一種簡單的方案可以解決上述問題： 對參與z值計算的所有維度值做全局Rank，用Rank值代替其原始值參與到z值計算中，由于Rank值一定是從0開始的正整數，完全符合z值構建條件，較好的解決上述問題。 在實驗中我們發現這種用Rank值的方法確實很有效，但是z值生成效率極低，計算引擎做全局Rank的代價是非常高的，基于Rank的方法效率瓶頸在于要做全局Rank計算，那么我們可不可以對原始數據做采樣減少數據量，用采樣后的數據計算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步

• 對輸入數據的key做sampling來估計key的分布，按指定的分區數切分成range并排序。計算出來的rangeBounds是一個長度為numPartition - 1 的數組，該數組里面每個元素表示一個分區內key值的上界/下界。

• shuffle write 過程中，每個輸入的key應該分到哪個分區內，由第一步計算出來的rangeBounds來確定。每個分區內的數據雖然沒有排序，但是注意rangeBounds是有序的因此分區之間宏觀上看是有序的，故只需對每個分區內數據做好排序即可保證數據全局有序。

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

• 對每個參與Z-Order的字段篩選規定個數（類比分區數）的Range并對進行排序，并計算出每個字段的RangeBounds；

• 實際映射過程中每個字段映射為該數據所在rangeBounds的中的下標，然后參與z-value的計算。可以看出由于區間下標是從0開始遞增的正整數，完全滿足z值生成條件；并且String類型的字段映射問題也被一并解決了。基于RangeBounds的z值生成方法，很好的解決了第一種方法所面臨的缺陷。由于多了一步采樣生成RangeBounds的過程，其效率顯然不如第一種方案，我們實現了上述兩種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結合

與Hudi的結合大致分為兩部分

3.2.1 表數據的Z排序重組

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

3.2.2 收集保存統計信息

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

• 讀取Parquet文件收集統計信息

/** 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方式收集統計信息

/** 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 應用到Spark查詢

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

• 將索引表加載到 IndexDataFrame

• 使用原始查詢過濾器為 IndexDataFrame 構建數據過濾器

• 查詢 IndexDataFrame 選擇候選文件

• 使用這些候選文件來重建 HudiMemoryIndex

通過min/max值和null計數信息為 IndexDataFrame 構建數據過濾器，由于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. 測試結果

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

“Z-Order加速Hudi大規模數據集的方法”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！