作者：vivo 網際網路搜尋團隊- Deng Jie

隨著技術的不斷的發展，巨量資料領域對於海量資料的儲存和處理的技術框架越來越多。在離線資料處理生態系統最具代表性的分散式處理引擎當屬Hive和Spark，它們在分割區策略方面有著一些相似之處，但也存在一些不同之處。

一、概述

隨著技術的不斷的發展，巨量資料領域對於海量資料的儲存和處理的技術框架越來越多。在離線資料處理生態系統最具代表性的分散式處理引擎當屬Hive和Spark，它們在分割區策略方面有著一些相似之處，但也存在一些不同之處。本篇文章將分析Hive與Spark分割區策略的異同點、它們各自的優缺點，以及一些優化措施。

二、Hive和Spark分割區概念

在瞭解Hive和Spark分割區內容之前，首先，我們先來回顧一下Hive和Spark的分割區概念。在Hive中，分割區是指將表中的資料劃分為不同的目錄或者子目錄，這些目錄或子目錄的名稱通常與表的列名相關聯。比如，一個名為「t_orders_name」的表可以按照日期分為多個目錄，每個目錄名稱對應一個日期值。這樣做的好處是可以大大提高查詢效率，因為只有涉及到特定日期的查詢才需要掃描對應的目錄，而不需要去掃描整個表。Spark的分割區概念與Hive類似，但是有一些不同之處，我們將在後文中進行討論。

在Hive中，分割區可以基於多個列進行，這些列的值組合形成目錄名稱。例如，如果我們將「t_orders_name」表按照日期和地區分割區，那麼目錄的名稱將包含日期和地區值的組合。在Hive中，資料儲存在分割區的目錄下，而不是儲存在表的目錄下。這使得Hive可以快速存取需要的資料，而不必掃描整個表。另外，Hive的分割區概念也可以用於資料分桶，分桶是將表中的資料劃分為固定數量的桶，每個桶包含相同的行。

而與Hive不同的是，Spark的分割區是將資料分成小塊以便平行計算處理。在Spark中，分割區的數量由Spark執行引擎根據資料大小和硬體資源自動計算得出。Spark的分割區數越多，可以並行處理的資料也就越多，因此也能更快的完成計算任務。但是，如果分割區數太多，將會導致過多的任務排程和資料傳輸開銷，從而降低整體的效能。因此，Spark分割區數的選擇應該考慮資料大小、硬體資源和計算任務複雜度等因素。

三、Hive和Spark分割區的應用場景

在瞭解Hive和Spark的分割區概念之後，接下來，我們來看看Hive和Spark分割區在不同的應用場景中有哪些不同的優勢。

3.1 Hive分割區

Hive分割區適用於巨量資料場景，可以對資料進行多級分割區，以便更細粒度地劃分資料，提高查詢效率。例如，在遊戲平臺的充值資料中，可以按照道具購買日期、道具付款狀態、遊戲使用者ID等多個維度進行分割區。這樣可以方便的進行資料統計、分析和查詢操作，同時避免單一分割區資料過大導致的效能問題。

3.2 Spark分割區

Spark分割區適用於大規模資料處理場景，可以充分利用叢集資源進行平行計算處理。比如，在機器學習演演算法的訓練過程中，可以將大量資料進行分割區，然後並行處理每個分割區的資料，從而提高演演算法的訓練速度和效率。另外，Spark的分散式計算引擎也可以支援在多個節點上進行資料分割區和計算，從而提高整個叢集的計算能力和效率。

簡而言之，Hive和Spark分割區在巨量資料處理和分散式計算場景這都有廣泛的應用，可以通過選擇合適的分割區策略和優化措施，進一步提高資料處理的效率和效能。

四、如何選擇分割區策略

在熟悉了Hive和Spark的分割區概念以及應用場景後。接下來，我們來看看在Hive和Spark中如何選擇分割區策略。分割區策略的選擇對資料處理的效率和效能有著重要的影響。下面將分別闡述Hive和Spark分割區策略的優缺點以及如何選擇分割區策略。

4.1 Hive分割區策略

優點：

• Hive的分割區策略可以提高查詢效率和資料處理效能，特別是在巨量資料集上表現突出。另外，Hive還支援多級分割區，允許更細粒度的資料劃分。

缺點：

• 在Hive中，分割區是以目錄的形式存在的，這會導致大量的目錄和子目錄，如果分割區過多，將會佔用過多的儲存空間。此外，Hive的分割區策略需要在建立表時進行設定，如果資料分佈出現變化，需要重新設定分割區策略。

4.2 Spark分割區策略

優點：

• Spark的分割區策略可以根據資料大小和硬體資源自動計算分割區數，這使得計算任務可以平行計算處理，從而提高了處理效率和效能。

缺點：

• 如果分割區數設定不當，將會導致過多的任務排程和資料傳輸開銷，從而影響整體效能。此外，Spark的分割區策略也需要根據資料大小、硬體資源和計算任務複雜度等因素進行調整。

4.3 分割區策略選擇

在實際專案開發使用中，選擇合適的分割區策略可以顯著提高資料處理的效率和效能。但是，如何選擇分割區策略需要根據具體情況進行考慮，這裡總結了一些分割區策略選擇的場景：

• 資料集大小：如果資料集較大，可以考慮使用Hive的多級劃分策略，以便更細粒度的劃分資料，提高查詢效率。如果資料集較小，可以使用Spark自動計算分割區策略，以便充分利用硬體資源並提高計算效率。
• 計算任務複雜度：如果計算任務比較複雜，例如需要進行多個JOIN操作，可以使用Hive的分桶策略，以便加快資料存取速度，減少JOIN操作的開銷。
• 硬體資源：分割區策略的選擇也需要考慮硬體資源的限制。如果硬體資源比較充足，可以增加分割區數以提高計算效率。如果硬體資源比較緊張，需要減少分割區數以避免任務排程和資料傳輸的開銷。

綜上所述，選擇合適的分割區策略需要根據具體的情況進行考慮，包括資料集大小、計算任務複雜度和硬體資源等因素。在實際使用中，可以通過實驗和偵錯來找到最佳的分割區策略。

五、如何優化分割區效能

除了選擇合適的分割區策略之外，還可以通過一些優化措施來進一步提高分割區的效能。在Spark中，大多數的Spark任務可以通過三個階段來表述，它們分別是讀取輸入資料、使用Spark處理、保持輸出資料。Spark雖然實際資料處理主要發生在記憶體中，但是Spark使用的是儲存在HDFS上的資料來作為輸入和輸出，任務的排程執行會使用大量的 I/O，存在效能瓶頸。

而Hive分割區資料是儲存在HDFS上的，然而HDFS對於大量小檔案支援不太友好，因為在每個NameNode記憶體中每個檔案大概有150位元組的儲存開銷，而整個HDFS叢集的IOPS數量是有上限的。當檔案寫入達到峰值時，會對HDFS叢集的基礎架構的某些部分產生效能瓶頸。

5.1 通過減少 I/O 頻寬來優化效能

在Hadoop叢集中，它依靠大規模並行 I/O 來支援數千個並行任務。比如現有一個大小為96TB的資料節點，磁碟的大小有兩種，它們分別是8TB和16TB。具有8TB磁碟的資料節點有12塊這樣的磁碟，而具有16TB磁碟的資料節點有6塊這樣的磁碟。我們可以假設每個磁碟的平均讀寫吞吐量約為100MB/s，而這兩種不同的磁碟分佈，它們對應的頻寬和IOPS，具體詳情如下表所示：

5.2 通過設定引數來優化效能

在Hadoop叢集中，每個資料節點為每個卷執行一個卷掃描器，用於掃描塊的狀態。由於卷掃描器與應用程式競爭磁碟資源，因此限制其磁碟頻寬很重要。設定 dfs.block.scanner.volume.bytes.per.second 屬性值來定義卷掃描器每秒可以掃描的位元組數，預設為1MB/s。

比如設定頻寬為5MB/s，掃描12TB所需要的時間為：12TB / 5MBps = (12 * 1024 * 1024 / (3600 * 24)) = 29.13天。

5.3 通過優化Spark處理分割區任務來提升效能

假如，現在需要重新計算曆史分割區的資料表，這種場景通常用於修復錯誤或者資料質量問題。在處理包含一年資料的大型資料集（比如1TB以上）時，可能會將資料分成幾千個Spark分割區來進行處理。雖然，從表面上看，這種處理方法並不是最合適的，使用動態分割區並將資料結果寫入按照日期分割區的Hive表中將產生多達上百萬個檔案。

下面，我們將任務分割區數縮小，現有一個包含3個分割區的Spark任務，並且想將資料寫入到包含3個分割區的Hive表。在這種情況下，希望傳送的是將3個檔案寫入到HDFS中，所有資料都儲存在每個分割區的單個檔案中。最終會生成9個檔案，並且每個檔案都有1個記錄。使用動態分割區寫入Hive表時，每個Spark分割區都由執行程式來並行處理。

處理Spark分割區資料時，每次執行程式在給定的Spark分割區中遇到新的分割區時，它都會開啟一個新檔案。預設情況下，Spark對資料會使用Hash或者Round Robin分割區器。當應用於任意資料時，可以假設這兩種方法在整個Spark分割區中相對均勻且隨機分佈資料。如下圖所示：

理想情況下，目標檔案大小應該大約是HDFS塊大小的倍數，預設情況下是128MB。在Hive中，提供了一些設定引數來自動將結果寫入到合理大小的檔案中，從開發者的角度來看幾乎是透明的，比如設定屬性 hive.merge.smallfiles.avgsize 和hive.merge.size.per.task 。但是，Spark中不存在此類功能，因此，我們需要自己開發實現，來確定一個資料集，應該寫入多少檔案。

5.3.1 基於大小的計算

理論上，這是最直接的方法，設定目標大小，估算資料的大小，然後進行劃分。但是，在很多情況下，檔案被寫入磁碟時會進行壓縮，並且其格式與儲存在 Java 堆中的記錄格式有所不同。這意味著估算寫入磁碟時記憶體的記錄大小不是一件容易的事情。雖然可以使用 Spark SizeEstimator應用程式通過記憶體中的資料的大小進行估算。但是，SizeEstimator會考慮資料框、資料集的內部消耗，以及資料的大小。總體來說，這種方式不太容易準確實現。

5.3.2 基於行數的計算

這種方法是設定目標行數，計算資料集的大小，然後執行除法來估算目標。我們的目標行數可以通過多種方式確定，或者通過為所有資料集選擇一個靜態數位，或者通過確定磁碟上單個記錄的大小並執行必要的計算。哪種方式最優，取決於你的資料集數量及其複雜性。計算相對來說成本較低，但是需要在計算前快取以避免重新計算資料集。

5.3.3 靜態檔案計算

最簡單的解決方案是，只要求開發者在每個寫入任務的基礎上，告訴Spark總共應該寫入多少個檔案。這種方式需要給開發者一些其他方法來獲取具體的數位，可以通過這種方式來替代昂貴的計算。

5.4. 優化Spark分發資料方式來提升效能

即使我們知道了如何將檔案寫入磁碟，但是，我們仍須讓Spark以符合實際的方式來構建我們的分割區。在Spark中，它提供了許多工具來確定資料在整個分割區中的分佈方式。但是，各種功能中隱藏著很多複雜性，在某些情況下，它們的含義並不明顯，下面將介紹Spark提供的一些選項來控制Spark輸出檔案的數量。

5.4.1 合併

Spark Coalesce是一個特殊版本的重新分割區，它只允許減少總的分割區，但是不需要完全的Shuffle，因此比重新分割區要快得多。它通過有效的合併分割區來實現這一點。如下圖所示：

Coalesce在某些情況下看起來是不錯的，但是也有一些問題。首先，Coalesce有一個難以使用的行為，以一個非常基礎的Spark應用程式為例，程式碼如下所示：

• Spark

load().map(…).filter(…).save()

比如，設定的並行度為1000，但是最終只想寫入10個檔案，可以設定如下：

• Spark

load().map(…).filter(…).coalesce(10).save()

但是，Spark會盡可能早的有效的將合併操作下推，因此這將執行為如下程式碼：

• Spark

load().coalesce(10).map(…).filter(…).save()

有效的解決這種問題的方法是在轉換和合並之間強制執行，程式碼如下所示：

• Spark

val df = load().map(…).filter(…).cache()
df.count()
df.coalesce(10)

在Spark中，快取是必須的，否則，你將不得不重新計算資料，這可能會重新消耗計算資源。然後，快取是需要消費一定資源的，如果你的資料集無法放入記憶體中，或者無法釋放記憶體，將資料有效的儲存在記憶體中兩次，那麼必須使用磁碟快取，這有其自身的侷限性和顯著的效能損失。

此外，正如我們看到的，通常需要執行Shuffle來獲得我們想要的更復雜的資料集結果。因此，Coalesce僅適用於特定的情況，比如如下場景：

• 保證只寫入一個Hive分割區；

• 目標檔案數少於你用於處理資料的Spark分割區數；

• 有充足的快取資源。

5.4.2 簡單重新分割區

在Spark中，一個簡單的重新分割區，可以通過設定引數來實現，比如df.repartition(100)。在這種情況下，使用迴圈分割區器，這意味著唯一的保證是輸出資料具有大致相同大小的Spark分割區，這種分割區僅適用於以下情況：

• 保證只需要寫入一個Hive分割區；

• 正在寫入的檔案數大於你的Spark分割區數，或者由於某些原因你無法使用合併。

5.4.3 按列重新分割區

按列重新分割區接收目標Spark分割區計數，以及要重新分割區的列序列，例如，df.repartition(100,$"date")。這對於強制要求Spark將具有相同鍵的資料，分發到同一個分割區很有用。一般來說，這對許多Spark操作（比如JOIN）很有用。

按列重新分割區使用HashPartitioner，將具有相同值的資料，分發給同一個分割區，實際上，它將執行以下操作：

但是，這種方法只有在每個分割區鍵都可以安全的寫入到一個檔案時才有效。這是因為無論有多少特定的Hash值，它們最終都會在同一個分割區中。按列重新分割區僅在你寫入一個或者多個小的Hive分割區時才有效。在任何其他情況下，它都是無效的，因為每個Hive分割區最終都會生成一個檔案，僅適用於最小的資料集。

5.4.4 按具有隨機因子的列重新分割區

我們可以通過新增約束的隨機因子來按列修改重新分割區，具體程式碼如下：

• Spark

df
.withColumn("rand", rand() % filesPerPartitionKey)
.repartition(100, $"key", $"rand")

理論上，只要滿足以下條件，這種方法應該會產生排序規則的資料和大小均勻的檔案：

• Hive分割區的大小大致相同；

• 知道每個Hive分割區的目標檔案數並且可以在執行時對其進行編碼。

但是，即使我們滿足上述這些條件，還有另外一個問題：雜湊衝突。假設，現在正在處理一年的資料，日期作為分割區的唯一鍵。如果每個分割區需要5個檔案，可以執行如下程式碼操作：

• Spark

df.withColumn("rand", rand() % 5).repartition(5*365, $"date", $"rand")

在後臺，Scala將構造一個包含日期和隨機因子的鍵，例如（,<0-4>）。然後，如果我們檢視HashPartitioner程式碼，可以發現它將執行以下操作：

• Spark

class HashPartitioner(partitions: Int) extends Partitioner {
    def getPartition(key: Any): Int = key match {
        case null => 0
        case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
    }
}

實際上，這裡面所做的事情，就是獲取關鍵元組的雜湊，然後使用目標數量的Spark分割區獲取它的mod。我們可以分析一下在這種情況下我們的資料將如何實現分佈，具體程式碼如下：

• Spark

import java.time.LocalDate
 
def hashCodeTuple(one: String, two: Int, mod: Int): Int = {
 val rawMod = (one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
def hashCodeSeq(one: String, two: Int, mod: Int): Int = {
 val rawMod = Seq(one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def iteration(numberDS: Int, filesPerPartition: Int): (Double, Double, Double) = {
  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(
    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, filesPerPartition*numberDS))
  )
 
  hashedRandKeys.size // Number of unique keys, with the random factor
 
  val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq
 
  groupedHashedKeys.size // number of actual sPartitions used
 
  val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse
   
  val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse
 
  sortedKeyCollisions.size // number of sPartitions with a hashing collision
 
  // (collisions, occurences)
  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse
   
  (
    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,
    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,
  sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble
  )
}
 
val results = Seq(
  iteration(365, 1),
  iteration(365, 5),
  iteration(365, 10),
  iteration(365, 100),
  iteration(365 * 2, 100),
  iteration(365 * 5, 100),
  iteration(365 * 10, 100)
)
 
val avgEfficiency = results.map(_._1).sum / results.length
val avgCollisionRate = results.map(_._2).sum / results.length
val avgSevereCollisionRate = results.map(_._3).sum / results.length
 
(avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2%, 42%, 12.6%

上面的指令碼計算了3個數量：

• 效率：非空的Spark分割區與輸出檔案數量的比率；

• 碰撞率：(date,rand)的Hash值傳送衝突的Spark分割區的百分比；

• 嚴重衝突率：同上，但是此鍵上的衝突次數為3或者更多。

衝突很重要，因為它們意味著我們的Spark分割區包含多個唯一的分割區鍵，而我們預計每個Spark分割區只有1個。我們從分析的結果可知，我們使用了63%的執行器，並且可能會出現嚴重的偏差，我們將近一半的執行正在處理比預期多2到3倍或者在某些情況下高達8倍的資料。

現在，有一個解決方法，即分割區縮放。在之前範例中，輸出的Spark分割區數量等於預期的總檔案數。如果將N個物件隨機分配給N個插槽，可以預期會有多個插槽包含多個物件，並且有幾個空插槽。因此，需要解決此問題，必須要降低物件與插槽的比率。

我們通過縮放輸出分割區計數來實現這一點，通過將輸出Spark分割區數乘以一個大因子，類似於：

• Spark

df
.withColumn("rand", rand() % 5)
.repartition(5*365*SCALING_FACTOR, $"date", $"rand")

具體分析程式碼如下所示：

• Spark

import java.time.LocalDate
 
def hashCodeTuple(one: String, two: Int, mod: Int): Int = {
 val rawMod = (one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def hashCodeSeq(one: String, two: Int, mod: Int): Int = {
 val rawMod = Seq(one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def iteration(numberDS: Int, filesPerPartition: Int, partitionFactor: Int = 1): (Double, Double, Double, Double) = {
  val partitionCount = filesPerPartition*numberDS * partitionFactor
  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(
    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, partitionCount))
  )
   
  hashedRandKeys.size // Number of unique keys, with the random factor
 
  val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq
 
  groupedHashedKeys.size // number of unique hashes - and thus, sPartitions with > 0 records
   
  val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse
   
  val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse
 
  sortedKeyCollisions.size // number of sPartitions with a hashing collision
 
  // (collisions, occurences)
  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse
   
  (
    groupedHashedKeys.size.toDouble / partitionCount,
    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,
    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,
    sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble
  )
}
 
// With a scale factor of 1
val results = Seq(
  iteration(365, 1),
  iteration(365, 5),
  iteration(365, 10),
  iteration(365, 100),
  iteration(365 * 2, 100),
  iteration(365 * 5, 100),
  iteration(365 * 10, 100)
)
 
val avgEfficiency = results.map(_._2).sum / results.length // What is the ratio of executors / output files
val avgCollisionRate = results.map(_._3).sum / results.length // What is the average collision rate
val avgSevereCollisionRate = results.map(_._4).sum / results.length // What is the average collision rate where 3 or more hashes collide
 
(avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2% Efficiency, 42% collision rate, 12.6% severe collision rate
 
iteration(365, 5, 2) // 37.7% partitions in-use, 77.4% Efficiency, 24.4% collision rate, 4.2% severe collision rate
iteration(365, 5, 5)
iteration(365, 5, 10)
iteration(365, 5, 100)

隨著我們的比例因子接近無窮大，碰撞很快接近於0，效率接近100%。但是，這會產生另外一個問題，即大量Spark分割區輸出將為空。同時這些空的Spark分割區也會帶來一些資源開銷，增加Driver的記憶體大小，會使我們更容易遇到，由於異常錯誤而導致分割區鍵空間意外增大的問題。

這裡的一個常見方法，是在使用這種方法時不顯示設定分割區（預設並行度和縮放），如果不提供分割區計數，則依賴Spark預設的spark.default.parallelism值。雖然，通常並行度自然高於總輸出檔案數（因此，隱式提供大於1 的縮放因子）。如果滿足以下條件，這種方式依然是一種有效的方法：

• Hive分割區的檔案數大致相等；

• 可以確定平均分割區檔案數應該是多少；

• 大致知道唯一分割區鍵的總數。

5.4.5 按範圍重新分割區

按範圍重新分割區是一個特列，它不使用RoundRobin和Hash Partitioner，而是使用一種特殊的方法，叫做Range Partitioner。

範圍分割區器根據某些給定鍵的順序在Spark分割區之間進行拆分行，但是，它不僅僅是全域性排序，而且還擁有以下特性：

• 具有相同雜湊的所有記錄將在同一個分割區中結束；

• 所有Spark分割區都將有一個最小值和最大值與之關聯；

• 最小值和最大值將通過使用取樣來檢測關鍵頻率和範圍來確定，分割區邊界將根據這些估計值進行初始設定；

• 分割區的大小不能保證完全相等，它們的相等性基於樣本的準確性，因此，預測的每個Spark分割區的最小值和最大值，分割區將根據需要增大或縮小來保證前兩個條件。

總而言之，範圍分割區將導致Spark建立與請求的Spark分割區數量相等的Bucket數量，然後它將這些Bucket對映到指定分割區鍵的範圍。例如，如果你的分割區鍵是日期，則範圍可能是（最小值2022-01-01，最大值2023-01-01）。然後，對於每條記錄，將記錄的分割區鍵與儲存Bucket的最小值和最大值進行比較，並相應的進行分配。如下圖所示：

六、總結

在選擇分割區策略時，需要根據具體的應用場景和需求進行選擇。常見的分割區策略包括按照時間、地域、使用者ID等多個維度進行分割區。在應用分割區策略時，還可以通過一些優化措施來進一步提高分割區的效能和效率，例如合理設定分割區數、避免過多的分割區列、減少重複資料等。

總之，分割區是巨量資料處理和分散式計算中非常重要的技術，可以幫助我們更好的管理和處理大規模的資料，提高資料處理的效率和效能，進而幫助我們更好的應對資料分析和業務應用的挑戰。

參考：

1. https://github.com/apache/spark

2. https://github.com/apache/hive

3. https://spark.apache.org/

4. https://hive.apache.org/