詳解 Apache Hudi Schema Evolution(模式演進)

2022 年 7 月 24 日
筆記

Schema Evolution（模式演進）允許用戶輕鬆更改 Hudi 表的當前模式，以適應隨時間變化的數據。從 0.11.0 版本開始，支援 Spark SQL（spark3.1.x 和 spark3.2.1）對 Schema 演進的 DDL 支援並且標誌為實驗性的。

場景

可以添加、刪除、修改和移動列（包括嵌套列）
分區列不能演進
不能對 Array 類型的嵌套列進行添加、刪除或操作

SparkSQL模式演進以及語法描述

使用模式演進之前，請先設置spark.sql.extensions，對於spark 3.2.x，需要設置spark.sql.catalog.spark_catalog

# Spark SQL for spark 3.1.x
spark-sql --packages org.apache.hudi:hudi-spark3.1.2-bundle_2.12:0.11.1,org.apache.spark:spark-avro_2.12:3.1.2 \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

# Spark SQL for spark 3.2.1
spark-sql --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.11.1,org.apache.spark:spark-avro_2.12:3.2.1 \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

啟動spark app後，請執行set schema.on.read.enable=true開啟模式演進

當前模式演進開啟後不能關閉

添加列

語法

-- add columns
ALTER TABLE Table name ADD COLUMNS(col_spec[, col_spec ...])

參數描述

參數	描述
tableName	表名
col_spec	列定義，由五個欄位組成，col_name, col_type, nullable, comment, col_position

col_name : 新列名，強制必須存在，如果在嵌套類型中添加子列，請指定子列的全路徑

示例

在嵌套類型users struct<name: string, age int>中添加子列col1，設置欄位為users.col1
在嵌套map類型member map<string, struct<n: string, a: int>>中添加子列col1, 設置欄位為member.value.col1

col_type : 新列的類型
nullable : 新列是否可為null，可為空，當前Hudi中並未使用
comment : 新列的注釋，可為空
col_position : 列添加的位置，值可為FIRST或者AFTER 某欄位

如果設置為FIRST，那麼新加的列在表的第一列
如果設置為AFTER 某欄位，將在某欄位後添加新列
如果設置為空，只有當新的子列被添加到嵌套列時，才能使用 FIRST。不要在頂級列中使用 FIRST。 AFTER 的使用沒有限制。

示例

alter table h0 add columns(ext0 string);
alter table h0 add columns(new_col int not null comment 'add new column' after col1);
alter table complex_table add columns(col_struct.col_name string comment 'add new column to a struct col' after col_from_col_struct);

修改列

語法

-- alter table ... alter column
ALTER TABLE Table name ALTER [COLUMN] col_old_name TYPE column_type [COMMENT] col_comment[FIRST|AFTER] column_name

參數描述

參數	描述
tableName	表名
col_old_name	待修改的列名
column_type	新的列類型
col_comment	列comment
column_name	列名，放置目標列的新位置。例如，AFTER column_name 表示目標列放在 column_name 之後

示例

--- Changing the column type
ALTER TABLE table1 ALTER COLUMN a.b.c TYPE bigint

--- Altering other attributes
ALTER TABLE table1 ALTER COLUMN a.b.c COMMENT 'new comment'
ALTER TABLE table1 ALTER COLUMN a.b.c FIRST
ALTER TABLE table1 ALTER COLUMN a.b.c AFTER x
ALTER TABLE table1 ALTER COLUMN a.b.c DROP NOT NULL

列類型變更矩陣表

源列類型\目標列類型	long	float	double	string	decimal	date	int
int	Y	Y	Y	Y	Y	N	Y
long	Y	N	Y	Y	Y	N	N
float	N	Y	Y	Y	Y	N	N
double	N	N	Y	Y	Y	N	N
decimal	N	N	N	Y	Y	N	N
string	N	N	N	Y	Y	Y	N
date	N	N	N	Y	N	Y	N

刪除列

語法

-- alter table ... drop columns
ALTER TABLE tableName DROP COLUMN|COLUMNS cols

示例

ALTER TABLE table1 DROP COLUMN a.b.c
ALTER TABLE table1 DROP COLUMNS a.b.c, x, y

修改列名

語法

-- alter table ... rename column
ALTER TABLE tableName RENAME COLUMN old_columnName TO new_columnName

示例

ALTER TABLE table1 RENAME COLUMN a.b.c TO x

修改表屬性

語法

-- alter table ... set|unset
ALTER TABLE Table name SET|UNSET tblproperties

示例

ALTER TABLE table SET TBLPROPERTIES ('table_property' = 'property_value')
ALTER TABLE table UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key')

修改表名

語法

-- alter table ... rename
ALTER TABLE tableName RENAME TO newTableName

示例

ALTER TABLE table1 RENAME TO table2

0.11.0之前的模式演進

模式演進是數據管理的一個非常重要的方面。 Hudi 支援開箱即用的常見模式演進場景，例如添加可為空的欄位或提升欄位的數據類型。此外，演進後的模式可以跨引擎查詢，例如 Presto、Hive 和 Spark SQL。下表總結了與不同 Hudi 表類型兼容的Schema變更類型。

Schema變更	COW	MOR	說明
在最後的根級別添加一個新的可為空列	Yes	Yes	Yes意味著具有演進模式的寫入成功並且寫入之後的讀取成功讀取整個數據集
向內部結構添加一個新的可為空列（最後）	Yes	Yes
添加具有默認值的新複雜類型欄位（map和array）	Yes	Yes
添加新的可為空列並更改欄位的順序	No	No	如果使用演進模式的寫入僅更新了一些基本文件而不是全部，則寫入成功但讀取失敗。目前Hudi 不維護模式註冊表，其中包含跨基礎文件的更改歷史記錄。然而如果 upsert 觸及所有基本文件，則讀取將成功
添加自定義可為空的 Hudi 元列，例如 _hoodie_meta_col	Yes	Yes
將根級別欄位的數據類型從 int 提升為 long	Yes	Yes	對於其他類型，Hudi 支援與Avro相同 Avro schema resolution
.
將嵌套欄位的數據類型從 int 提升為 long	Yes	Yes
對於複雜類型（map或array的值），將數據類型從 int 提升為 long	Yes	Yes
在最後的根級別添加一個新的不可為空的列	No	No	對於Spark數據源的MOR表，寫入成功但讀取失敗。作為一種解決方法，您可以使該欄位為空
向內部結構添加一個新的不可為空的列（最後）	No	No
將嵌套欄位的數據類型從 long 更改為 int	No	No
將複雜類型的數據類型從 long 更改為 int（映射或數組的值）	No	No

讓我們通過一個示例來演示 Hudi 中的模式演進支援。在下面的示例中，我們將添加一個新的字元串欄位並將欄位的數據類型從 int 更改為 long。

Welcome to
    ____              __
    / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
    /___/ .__/\_,_/_/ /_/\_\   version 3.1.2
    /_/

    Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_292)
    Type in expressions to have them evaluated.
    Type :help for more information.

scala> import org.apache.hudi.QuickstartUtils._
import org.apache.hudi.QuickstartUtils._

scala> import scala.collection.JavaConversions._
import scala.collection.JavaConversions._

scala> import org.apache.spark.sql.SaveMode._
import org.apache.spark.sql.SaveMode._

scala> import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceReadOptions._

scala> import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.DataSourceWriteOptions._

scala> import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.config.HoodieWriteConfig._

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row

scala> val tableName = "hudi_trips_cow"
    tableName: String = hudi_trips_cow
scala> val basePath = "file:///tmp/hudi_trips_cow"
    basePath: String = file:///tmp/hudi_trips_cow
scala> val schema = StructType( Array(
    | StructField("rowId", StringType,true),
    | StructField("partitionId", StringType,true),
    | StructField("preComb", LongType,true),
    | StructField("name", StringType,true),
    | StructField("versionId", StringType,true),
    | StructField("intToLong", IntegerType,true)
    | ))
    schema: org.apache.spark.sql.types.StructType = StructType(StructField(rowId,StringType,true), StructField(partitionId,StringType,true), StructField(preComb,LongType,true), StructField(name,StringType,true), StructField(versionId,StringType,true), StructField(intToLong,IntegerType,true))
    
scala> val data1 = Seq(Row("row_1", "part_0", 0L, "bob", "v_0", 0),
    |                Row("row_2", "part_0", 0L, "john", "v_0", 0),
    |                Row("row_3", "part_0", 0L, "tom", "v_0", 0))
    data1: Seq[org.apache.spark.sql.Row] = List([row_1,part_0,0,bob,v_0,0], [row_2,part_0,0,john,v_0,0], [row_3,part_0,0,tom,v_0,0])

scala> var dfFromData1 = spark.createDataFrame(data1, schema)
scala> dfFromData1.write.format("hudi").
    |   options(getQuickstartWriteConfigs).
    |   option(PRECOMBINE_FIELD_OPT_KEY.key, "preComb").
    |   option(RECORDKEY_FIELD_OPT_KEY.key, "rowId").
    |   option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionId").
    |   option("hoodie.index.type","SIMPLE").
    |   option(TABLE_NAME.key, tableName).
    |   mode(Overwrite).
    |   save(basePath)

scala> var tripsSnapshotDF1 = spark.read.format("hudi").load(basePath + "/*/*")
    tripsSnapshotDF1: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 9 more fields]

scala> tripsSnapshotDF1.createOrReplaceTempView("hudi_trips_snapshot")

scala> spark.sql("desc hudi_trips_snapshot").show()
    +--------------------+---------+-------+
    |            col_name|data_type|comment|
    +--------------------+---------+-------+
    | _hoodie_commit_time|   string|   null|
    |_hoodie_commit_seqno|   string|   null|
    |  _hoodie_record_key|   string|   null|
    |_hoodie_partition...|   string|   null|
    |   _hoodie_file_name|   string|   null|
    |               rowId|   string|   null|
    |         partitionId|   string|   null|
    |             preComb|   bigint|   null|
    |                name|   string|   null|
    |           versionId|   string|   null|
    |           intToLong|      int|   null|
    +--------------------+---------+-------+
    
scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong from hudi_trips_snapshot").show()
    +-----+-----------+-------+----+---------+---------+
    |rowId|partitionId|preComb|name|versionId|intToLong|
    +-----+-----------+-------+----+---------+---------+
    |row_3|     part_0|      0| tom|      v_0|        0|
    |row_2|     part_0|      0|john|      v_0|        0|
    |row_1|     part_0|      0| bob|      v_0|        0|
    +-----+-----------+-------+----+---------+---------+

// In the new schema, we are going to add a String field and 
// change the datatype `intToLong` field from  int to long.
scala> val newSchema = StructType( Array(
    | StructField("rowId", StringType,true),
    | StructField("partitionId", StringType,true),
    | StructField("preComb", LongType,true),
    | StructField("name", StringType,true),
    | StructField("versionId", StringType,true),
    | StructField("intToLong", LongType,true),
    | StructField("newField", StringType,true)
    | ))
    newSchema: org.apache.spark.sql.types.StructType = StructType(StructField(rowId,StringType,true), StructField(partitionId,StringType,true), StructField(preComb,LongType,true), StructField(name,StringType,true), StructField(versionId,StringType,true), StructField(intToLong,LongType,true), StructField(newField,StringType,true))

scala> val data2 = Seq(Row("row_2", "part_0", 5L, "john", "v_3", 3L, "newField_1"),
    |                Row("row_5", "part_0", 5L, "maroon", "v_2", 2L, "newField_1"),
    |                Row("row_9", "part_0", 5L, "michael", "v_2", 2L, "newField_1"))
    data2: Seq[org.apache.spark.sql.Row] = List([row_2,part_0,5,john,v_3,3,newField_1], [row_5,part_0,5,maroon,v_2,2,newField_1], [row_9,part_0,5,michael,v_2,2,newField_1])

scala> var dfFromData2 = spark.createDataFrame(data2, newSchema)
scala> dfFromData2.write.format("hudi").
    |   options(getQuickstartWriteConfigs).
    |   option(PRECOMBINE_FIELD_OPT_KEY.key, "preComb").
    |   option(RECORDKEY_FIELD_OPT_KEY.key, "rowId").
    |   option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionId").
    |   option("hoodie.index.type","SIMPLE").
    |   option(TABLE_NAME.key, tableName).
    |   mode(Append).
    |   save(basePath)

scala> var tripsSnapshotDF2 = spark.read.format("hudi").load(basePath + "/*/*")
    tripsSnapshotDF2: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 10 more fields]

scala> tripsSnapshotDF2.createOrReplaceTempView("hudi_trips_snapshot")

scala> spark.sql("desc hudi_trips_snapshot").show()
    +--------------------+---------+-------+
    |            col_name|data_type|comment|
    +--------------------+---------+-------+
    | _hoodie_commit_time|   string|   null|
    |_hoodie_commit_seqno|   string|   null|
    |  _hoodie_record_key|   string|   null|
    |_hoodie_partition...|   string|   null|
    |   _hoodie_file_name|   string|   null|
    |               rowId|   string|   null|
    |         partitionId|   string|   null|
    |             preComb|   bigint|   null|
    |                name|   string|   null|
    |           versionId|   string|   null|
    |           intToLong|   bigint|   null|
    |            newField|   string|   null|
    +--------------------+---------+-------+


scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong, newField from hudi_trips_snapshot").show()
    +-----+-----------+-------+-------+---------+---------+----------+
    |rowId|partitionId|preComb|   name|versionId|intToLong|  newField|
    +-----+-----------+-------+-------+---------+---------+----------+
    |row_3|     part_0|      0|    tom|      v_0|        0|      null|
    |row_2|     part_0|      5|   john|      v_3|        3|newField_1|
    |row_1|     part_0|      0|    bob|      v_0|        0|      null|
    |row_5|     part_0|      5| maroon|      v_2|        2|newField_1|
    |row_9|     part_0|      5|michael|      v_2|        2|newField_1|
    +-----+-----------+-------+-------+---------+---------+----------+

Tags:

詳解 Apache Hudi Schema Evolution(模式演進)

場景

SparkSQL模式演進以及語法描述

添加列

修改列

刪除列

修改列名

修改表屬性

修改表名

0.11.0之前的模式演進

VirMach 便宜 VPS

QNews

詳解 Apache Hudi Schema Evolution(模式演進)

場景

SparkSQL模式演進以及語法描述

添加列

修改列

刪除列

修改列名

修改表屬性

修改表名

0.11.0之前的模式演進

分享此文：

Related Posts

ShardingSphere資料庫讀寫分離

Linux 啟動流程及相關知識

Linux 啟動流程及相關知識

最美公路變垃圾公路：遊客激增 基本素質不能丟

VirMach 便宜 VPS

QNews

熱門文章

熱門搜尋

最美公路變垃圾公路：遊客激增基本素質不能丟