使用Apache Spark和Apache Hudi構建分析數據湖

由 admin · 已發表 2020 年 11 月 30 日 · 已更新 2020 年 11 月 30 日

1. 引入

大多數現代數據湖都是基於某種分佈式文件系統（DFS），如HDFS或基於雲的存儲，如AWS S3構建的。遵循的基本原則之一是文件的“一次寫入多次讀取”訪問模型。這對於處理海量數據非常有用，如數百GB到TB的數據。

但是在構建分析數據湖時，更新數據並不罕見。根據不同場景，這些更新頻率可能是每小時一次，甚至可能是每天或每周一次。另外可能還需要在最新視圖、包含所有更新的歷史視圖甚至僅是最新增量視圖上運行分析。

通常這會導致使用用於流和批處理的多個系統，前者處理增量數據，而後者處理歷史數據。

處理存儲在HDFS上的數據時，維護增量更新的常見工作流程是這裏所述的Ingest-Reconcile-Compact-Purge策略。

Apache Hudi之類的框架在這裏便可發揮作用。它在後台為我們管理此工作流程，從而使我們的核心應用程序代碼更加簡潔，Hudi支持對最新數據視圖的查詢以及查詢在某個時間點的增量更改。

這篇文章將介紹Hudi的核心概念以及如何在Copy-On-Write模式下進行操作。

本篇文章項目源代碼放在github。

2. 大綱

先決條件和框架版本
Hudi核心概念
初始設置和依賴項
使用CoW表

2.1 先決條件和框架版本

如果你事先了解如何使用scala編寫spark作業以及讀取和寫入parquet文件，那麼本篇文章理解起來將非常容易。

框架版本如下

JDK: openjdk 1.8.0_242
Scala: 2.12.8
Spark: 2.4.4
Hudi Spark bundle: 0.5.2-incubating

注意：在撰寫本文時，AWS EMR與Hudi v0.5.0-incubating集成在一起，該軟件包具有一個bug會導致upsert操作卡死或花費很長時間才能完成，可查看相關issue了解更多，該問題已在當前版本的Hudi（0.5.2-incubating及之後版本）中修復。如果計劃在AWS EMR上運行代碼，則可能要考慮用最新版本覆蓋默認的集成版本。

2.2 Hudi核心概念

先從一些需要理解的核心概念開始。

1. 表類型

Hudi支持兩種表類型

寫時複製（CoW）：寫入CoW表時，將運行Ingest-Reconcile-Compact-Purge周期。每次寫操作后，CoW表中的數據始終是最新記錄，對於需要儘快讀取最新數據的場景，可首選此模式。數據僅以列文件格式（parquet）存儲在CoW表中，由於每個寫操作都涉及壓縮和覆蓋，因此此模式產生的文件最少。
讀時合併（MoR）：MoR表專註於快速寫操作。寫入這些表將創建增量文件，隨後將其壓縮以生成讀取時的最新數據，壓縮操作可以同步或異步完成，數據以列文件格式（parquet）和基於行的文件格式（avro）組合存儲。

這是Hudi文檔中提到的兩種表格格式之間的權衡取捨。

Trade-off	CoW	MoR
數據延遲	Higher	Lower
更新開銷 (I/O)	Higher (重寫整個parquet文件)	Lower (追加到delta log文件)
Parquet文件大小	Smaller (高update(I/0) 開銷)	Larger (低更新開銷)
Write Amplification	Higher	Lower (由compaction策略決定)

2. 查詢類型

Hudi支持兩種主要類型的查詢：“快照查詢”和“增量查詢”。除兩種主要查詢類型外，MoR表還支持“讀優化查詢”。

快照查詢：對於CoW表，快照查詢返回數據的最新視圖，而對於MoR表，則返回接近實時的視圖。對於MoR表，快照查詢將即時合併基本文件和增量文件，因此可能會有一些讀取延遲。使用CoW，由於寫入負責合併，因此讀取很快，只需要讀取基本文件。
增量查詢：增量查詢使您可以通過指定“開始”時間或在特定時間點通過指定“開始”和“結束”時間來查看特定提交時間之後的數據。
讀優化查詢：對於MoR表，讀取優化查詢返回一個視圖，該視圖僅包含基本文件中的數據，而不合併增量文件。

3. 以Hudi格式寫入時的關鍵屬性

hoodie.datasource.write.table.type，定義表的類型-默認值為COPY_ON_WRITE。對於MoR表，將此值設置為MERGE_ON_READ。
hoodie.table.name，這是必填字段，每個表都應具有唯一的名稱。
hoodie.datasource.write.recordkey.field，將此視為表的主鍵。此屬性的值是DataFrame中列的名稱，該列是主鍵。
hoodie.datasource.write.precombine.field，更新數據時，如果存在兩個具有相同主鍵的記錄，則此列中的值將決定更新哪個記錄。選擇諸如時間戳記的列將確保選擇具有最新時間戳記的記錄。
hoodie.datasource.write.operation，定義寫操作的類型。值可以為upsert，insert，bulk_insert和delete，默認值為upsert。

2.3 初始設置和依賴項

1. 依賴說明

為了在Spark作業中使用Hudi，需要使用spark-sql，hudi-spark-bundle和spark-avro依賴項，此外還需要將Spark配置為使用KryoSerializer。

pom.xml大致內容如下

<properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <encoding>UTF-8</encoding>
    <scala.version>2.12.8</scala.version>
    <scala.compat.version>2.12</scala.compat.version>
    <spec2.version>4.2.0</spec2.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>${scala.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_${scala.compat.version}</artifactId>
        <version>2.4.4</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hudi</groupId>
        <artifactId>hudi-spark-bundle_${scala.compat.version}</artifactId>
        <version>0.5.2-incubating</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-avro_${scala.compat.version}</artifactId>
        <version>2.4.4</version>
    </dependency>
</dependencies>

2. 設置Schema

我們使用下面的Album類來表示表的schema。

case class Album(albumId: Long, title: String, tracks: Array[String], updateDate: Long)

3. 生成測試數據

創建一些用於upsert操作的數據。

INITIAL_ALBUM_DATA有兩個記錄，鍵為801。
UPSERT_ALBUM_DATA包含一個更新的記錄和兩個新的記錄。

def dateToLong(dateString: String): Long = LocalDate.parse(dateString, formatter).toEpochDay

private val INITIAL_ALBUM_DATA = Seq(
    Album(800, "6 String Theory", Array("Lay it down", "Am I Wrong", "68"), dateToLong("2019-12-01")),
    Album(801, "Hail to the Thief", Array("2+2=5", "Backdrifts"), dateToLong("2019-12-01")),
    Album(801, "Hail to the Thief", Array("2+2=5", "Backdrifts", "Go to sleep"), dateToLong("2019-12-03"))
  )

  private val UPSERT_ALBUM_DATA = Seq(
    Album(800, "6 String Theory - Special", Array("Jumpin' the blues", "Bluesnote", "Birth of blues"), dateToLong("2020-01-03")),
    Album(802, "Best Of Jazz Blues", Array("Jumpin' the blues", "Bluesnote", "Birth of blues"), dateToLong("2020-01-04")),
    Album(803, "Birth of Cool", Array("Move", "Jeru", "Moon Dreams"), dateToLong("2020-02-03"))
  )

4. 初始化SparkContext

最後初始化Spark上下文。這裏要注意的重要一點是KryoSerializer的使用。

val spark: SparkSession = SparkSession.builder()
    .appName("hudi-datalake")
    .master("local[*]")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.hive.convertMetastoreParquet", "false") // Uses Hive SerDe, this is mandatory for MoR tables
    .getOrCreate()

2.4 使用CoW表

本節將處理CoW表的記錄，如讀取和刪除記錄。

1. basePath(基本路徑)和Upsert方法

定義一個basePath，upsert方法會將表數據寫入該路徑，該方法將以org.apache.hudi格式寫入Dataframe，請確保上面討論的所有Hudi屬性均已設置。

val basePath = "/tmp/store"

private def upsert(albumDf: DataFrame, tableName: String, key: String, combineKey: String) = {
    albumDf.write
      .format("hudi")
      .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
      .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, key)
      .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, combineKey)
      .option(HoodieWriteConfig.TABLE_NAME, tableName)
      .option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
      // Ignore this property for now, the default is too high when experimenting on your local machine
      // Set this to a lower value to improve performance.
      // I'll probably cover Hudi tuning in a separate post.
      .option("hoodie.upsert.shuffle.parallelism", "2")
      .mode(SaveMode.Append)
      .save(s"$basePath/$tableName/")
  }

2. 初始化upsert

插入INITIAL_ALBUM_DATA，我們應該創建2條記錄，對於801，該記錄的日期為2019-12-03。

val tableName = "Album"
upsert(INITIAL_ALBUM_DATA.toDF(), tableName, "albumId", "updateDate")
spark.read.format("hudi").load(s"$basePath/$tableName/*").show()

讀取CoW表就像使用格式（“hudl”）的常規spark.read一樣簡單。

// Output
+-------------------+--------------------+------------------+----------------------+--------------------+-------+-----------------+--------------------+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|albumId|            title|              tracks|updateDate|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+-----------------+--------------------+----------+
|     20200412182343|  20200412182343_0_1|               801|               default|65841d0a-0083-447...|    801|Hail to the Thief|[2+2=5, Backdrift...|     18233|
|     20200412182343|  20200412182343_0_2|               800|               default|65841d0a-0083-447...|    800|  6 String Theory|[Lay it down, Am ...|     18231|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+-----------------+--------------------+----------+

另一種確定的方法是查看Workload profile的日誌輸出，內容大致如下

Workload profile :WorkloadProfile {globalStat=WorkloadStat {numInserts=2, numUpdates=0}, partitionStat={default=WorkloadStat {numInserts=2, numUpdates=0}}}

3. 更新記錄

upsert(UPSERT_ALBUM_DATA.toDF(), tableName, "albumId", "updateDate")

查看Workload profile的日誌輸出，並驗證它是否符合預期

Workload profile :WorkloadProfile {globalStat=WorkloadStat {numInserts=2, numUpdates=1}, partitionStat={default=WorkloadStat {numInserts=2, numUpdates=1}}}

查詢輸出如下

spark.read.format("hudi").load(s"$basePath/$tableName/*").show()

//Output
+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|albumId|               title|              tracks|updateDate|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+
|     20200412183510|  20200412183510_0_1|               801|               default|65841d0a-0083-447...|    801|   Hail to the Thief|[2+2=5, Backdrift...|     18233|
|     20200412184040|  20200412184040_0_1|               800|               default|65841d0a-0083-447...|    800|6 String Theory -...|[Jumpin' the blue...|     18264|
|     20200412184040|  20200412184040_0_2|               802|               default|65841d0a-0083-447...|    802|  Best Of Jazz Blues|[Jumpin' the blue...|     18265|
|     20200412184040|  20200412184040_0_3|               803|               default|65841d0a-0083-447...|    803|       Birth of Cool|[Move, Jeru, Moon...|     18295|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+

4. 查詢記錄

我們在上面查看數據的方式稱為“快照查詢”，這是默認設置，另外還支持“增量查詢”。

4.1 增量查詢

要執行增量查詢，我們需要在讀取時將hoodie.datasource.query.type屬性設置為incremental，並指定hoodie.datasource.read.begin.instanttime屬性。這將在指定的即時時間之後讀取所有記錄，對於本示例，我們將instantTime指定為20200412183510。

spark.read
    .format("hudi")
    .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY, DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
    .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, "20200412183510")
    .load(s"$basePath/$tableName")
    .show()

這將在提交時間20200412183510之後返回所有記錄。

+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|albumId|               title|              tracks|updateDate|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+
|     20200412184040|  20200412184040_0_1|               800|               default|65841d0a-0083-447...|    800|6 String Theory -...|[Jumpin' the blue...|     18264|
|     20200412184040|  20200412184040_0_2|               802|               default|65841d0a-0083-447...|    802|  Best Of Jazz Blues|[Jumpin' the blue...|     18265|
|     20200412184040|  20200412184040_0_3|               803|               default|65841d0a-0083-447...|    803|       Birth of Cool|[Move, Jeru, Moon...|     18295|
+-------------------+--------------------+------------------+----------------------+--------------------+-------+--------------------+--------------------+----------+

5. 刪除記錄

我們要查看的最後一個操作是刪除，刪除類似於upsert，需要一個待刪除記錄的DataFrame，如下面的示例代碼所示，不需要整行，只需要主鍵即可。

val deleteKeys = Seq(
    Album(803, "", null, 0l),
    Album(802, "", null, 0l)
)

import spark.implicits._

val df = deleteKeys.toDF()

df.write.format("hudi")
    .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
    .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "albumId")
    .option(HoodieWriteConfig.TABLE_NAME, tableName)
    // Set the option "hoodie.datasource.write.operation" to "delete"
    .option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL)
    .mode(SaveMode.Append) // Only Append Mode is supported for Delete.
    .save(s"$basePath/$tableName/")

spark.read.format("hudi").load(s"$basePath/$tableName/*").show()

這是本部分介紹的全部內容。後面我們將探討在MERGE-ON-READ表進行操作。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※超省錢租車方案

※別再煩惱如何寫文案,掌握八大原則!

※回頭車貨運收費標準

※教你寫出一流的銷售文案?

※產品缺大量曝光嗎?你需要的是一流包裝設計!

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

使用Apache Spark和Apache Hudi構建分析數據湖

1. 引入

2. 大綱

2.1 先決條件和框架版本

2.2 Hudi核心概念

2.3 初始設置和依賴項

2.4 使用CoW表

您可能也會喜歡…

近期文章

分類

彙整

使用Apache Spark和Apache Hudi構建分析數據湖

1. 引入

2. 大綱

2.1 先決條件和框架版本

2.2 Hudi核心概念

2.3 初始設置和依賴項

2.4 使用CoW表

您可能也會喜歡…

【2023亞運圍棋賽程】9/28 中華隊 vs 中國，許皓鋐金牌決賽直播轉播 LIVE 線上看

全新 M3 MacBook Pro、iMac 通過 NCC 認證，有望於近期在台開賣

預計蘋果將在近期推出 iOS 17.0.3，修復 iPhone 15 Pro 過熱問題

近期文章

標籤

分類

彙整