图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › 数据湖 › 介绍解说› 详解Apache Hudi Schema Evolution(模式演进)
分享

详解Apache Hudi Schema Evolution(模式演进)

nettman 2022-6-27 23:38:04 发表于 介绍解说 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 3276
本帖最后由 nettman 于 2022-6-27 23:40 编辑

问题导读

1.SparkSQL模式演进需要设置哪个参数？
2.如何添加列？
3.如何修改列？


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

  1. # Spark SQL for spark 3.1.x

  2. 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 \

  3. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \

  4. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

  5. 

  6. # Spark SQL for spark 3.2.1

  7. spark-sql --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.11.1,org.apache.spark:spark-avro_2.12:3.2.1 \

  8. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \

  9. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \

  10. --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 ...])


参数描述


1.png

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 的使用没有限制。

示例
  1. alter table h0 add columns(ext0 string);

  2. alter table h0 add columns(new_col int not null comment 'add new column' after col1);

  3. alter table complex_table add columns(col_struct.col_name string comment 'add new column to a struct col' after col_from_col_struct);
复制代码

修改列

语法
  1. -- alter table ... alter column

  2. ALTER TABLE Table name ALTER [COLUMN] col_old_name TYPE column_type [COMMENT] col_comment[FIRST|AFTER] column_name
复制代码
参数描述

1.png

示例
  1. --- Changing the column type

  2. ALTER TABLE table1 ALTER COLUMN a.b.c TYPE bigint

  3. 

  4. --- Altering other attributes

  5. ALTER TABLE table1 ALTER COLUMN a.b.c COMMENT 'new comment'

  6. ALTER TABLE table1 ALTER COLUMN a.b.c FIRST

  7. ALTER TABLE table1 ALTER COLUMN a.b.c AFTER x

  8. ALTER TABLE table1 ALTER COLUMN a.b.c DROP NOT NULL
复制代码

列类型变更兼容表

1.png

删除列

语法

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

示例

  1. <div>ALTER TABLE table1 DROP COLUMN a.b.c</div><div>ALTER TABLE table1 DROP COLUMNS a.b.c, x, y</div>
复制代码

修改列名

语法

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

示例

  1. ALTER TABLE table1 RENAME COLUMN a.b.c TO x
复制代码

修改表属性

语法

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

示例
  1. ALTER TABLE table SET TBLPROPERTIES ('table_property' = 'property_value')</div><div>ALTER TABLE table UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key')
复制代码

修改表名

语法

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

示例
  1. ALTER TABLE table1 RENAME TO table2
复制代码

0.11.0之前的模式演进

模式演进是数据管理的一个非常重要的方面。Hudi 支持开箱即用的常见模式演进场景，例如添加可为空的字段或提升字段的数据类型。此外，演进后的模式可以跨引擎查询，例如 Presto、Hive 和 Spark SQL。下表总结了与不同 Hudi 表类型兼容的Schema变更类型。

1.png

让我们通过一个示例来演示 Hudi 中的模式演进支持。在下面的示例中，我们将添加一个新的字符串字段并将字段的数据类型从 int 更改为 long。

  1. Welcome to

  2.     ____              __

  3.     / __/__  ___ _____/ /__

  4.     _\ \/ _ \/ _ `/ __/  '_/

  5.     /___/ .__/\_,_/_/ /_/\_\   version 3.1.2

  6.     /_/

  7. 

  8.     Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_292)

  9.     Type in expressions to have them evaluated.

  10.     Type :help for more information.

  11. 

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

  13. import org.apache.hudi.QuickstartUtils._

  14. 

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

  16. import scala.collection.JavaConversions._

  17. 

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

  19. import org.apache.spark.sql.SaveMode._

  20. 

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

  22. import org.apache.hudi.DataSourceReadOptions._

  23. 

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

  25. import org.apache.hudi.DataSourceWriteOptions._

  26. 

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

  28. import org.apache.hudi.config.HoodieWriteConfig._

  29. 

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

  31. import org.apache.spark.sql.types._

  32. 

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

  34. import org.apache.spark.sql.Row

  35. 

  36. scala> val tableName = "hudi_trips_cow"

  37.     tableName: String = hudi_trips_cow

  38. scala> val basePath = "file:///tmp/hudi_trips_cow"

  39.     basePath: String = file:///tmp/hudi_trips_cow

  40. scala> val schema = StructType( Array(

  41.     | StructField("rowId", StringType,true),

  42.     | StructField("partitionId", StringType,true),

  43.     | StructField("preComb", LongType,true),

  44.     | StructField("name", StringType,true),

  45.     | StructField("versionId", StringType,true),

  46.     | StructField("intToLong", IntegerType,true)

  47.     | ))

  48.     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))

  49.     

  50. scala> val data1 = Seq(Row("row_1", "part_0", 0L, "bob", "v_0", 0),

  51.     |                Row("row_2", "part_0", 0L, "john", "v_0", 0),

  52.     |                Row("row_3", "part_0", 0L, "tom", "v_0", 0))

  53.     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])

  54. 

  55. scala> var dfFromData1 = spark.createDataFrame(data1, schema)

  56. scala> dfFromData1.write.format("hudi").

  57.     |   options(getQuickstartWriteConfigs).

  58.     |   option(PRECOMBINE_FIELD_OPT_KEY.key, "preComb").

  59.     |   option(RECORDKEY_FIELD_OPT_KEY.key, "rowId").

  60.     |   option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionId").

  61.     |   option("hoodie.index.type","SIMPLE").

  62.     |   option(TABLE_NAME.key, tableName).

  63.     |   mode(Overwrite).

  64.     |   save(basePath)

  65. 

  66. scala> var tripsSnapshotDF1 = spark.read.format("hudi").load(basePath + "/*/*")

  67.     tripsSnapshotDF1: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 9 more fields]

  68. 

  69. scala> tripsSnapshotDF1.createOrReplaceTempView("hudi_trips_snapshot")

  70. 

  71. scala> spark.sql("desc hudi_trips_snapshot").show()

  72.     +--------------------+---------+-------+

  73.     |            col_name|data_type|comment|

  74.     +--------------------+---------+-------+

  75.     | _hoodie_commit_time|   string|   null|

  76.     |_hoodie_commit_seqno|   string|   null|

  77.     |  _hoodie_record_key|   string|   null|

  78.     |_hoodie_partition...|   string|   null|

  79.     |   _hoodie_file_name|   string|   null|

  80.     |               rowId|   string|   null|

  81.     |         partitionId|   string|   null|

  82.     |             preComb|   bigint|   null|

  83.     |                name|   string|   null|

  84.     |           versionId|   string|   null|

  85.     |           intToLong|      int|   null|

  86.     +--------------------+---------+-------+

  87.     

  88. scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong from hudi_trips_snapshot").show()

  89.     +-----+-----------+-------+----+---------+---------+

  90.     |rowId|partitionId|preComb|name|versionId|intToLong|

  91.     +-----+-----------+-------+----+---------+---------+

  92.     |row_3|     part_0|      0| tom|      v_0|        0|

  93.     |row_2|     part_0|      0|john|      v_0|        0|

  94.     |row_1|     part_0|      0| bob|      v_0|        0|

  95.     +-----+-----------+-------+----+---------+---------+

  96. 

  97. // In the new schema, we are going to add a String field and 

  98. // change the datatype `intToLong` field from  int to long.

  99. scala> val newSchema = StructType( Array(

  100.     | StructField("rowId", StringType,true),

  101.     | StructField("partitionId", StringType,true),

  102.     | StructField("preComb", LongType,true),

  103.     | StructField("name", StringType,true),

  104.     | StructField("versionId", StringType,true),

  105.     | StructField("intToLong", LongType,true),

  106.     | StructField("newField", StringType,true)

  107.     | ))

  108.     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))

  109. 

  110. scala> val data2 = Seq(Row("row_2", "part_0", 5L, "john", "v_3", 3L, "newField_1"),

  111.     |                Row("row_5", "part_0", 5L, "maroon", "v_2", 2L, "newField_1"),

  112.     |                Row("row_9", "part_0", 5L, "michael", "v_2", 2L, "newField_1"))

  113.     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])

  114. 

  115. scala> var dfFromData2 = spark.createDataFrame(data2, newSchema)

  116. scala> dfFromData2.write.format("hudi").

  117.     |   options(getQuickstartWriteConfigs).

  118.     |   option(PRECOMBINE_FIELD_OPT_KEY.key, "preComb").

  119.     |   option(RECORDKEY_FIELD_OPT_KEY.key, "rowId").

  120.     |   option(PARTITIONPATH_FIELD_OPT_KEY.key, "partitionId").

  121.     |   option("hoodie.index.type","SIMPLE").

  122.     |   option(TABLE_NAME.key, tableName).

  123.     |   mode(Append).

  124.     |   save(basePath)

  125. 

  126. scala> var tripsSnapshotDF2 = spark.read.format("hudi").load(basePath + "/*/*")

  127.     tripsSnapshotDF2: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 10 more fields]

  128. 

  129. scala> tripsSnapshotDF2.createOrReplaceTempView("hudi_trips_snapshot")

  130. 

  131. scala> spark.sql("desc hudi_trips_snapshot").show()

  132.     +--------------------+---------+-------+

  133.     |            col_name|data_type|comment|

  134.     +--------------------+---------+-------+

  135.     | _hoodie_commit_time|   string|   null|

  136.     |_hoodie_commit_seqno|   string|   null|

  137.     |  _hoodie_record_key|   string|   null|

  138.     |_hoodie_partition...|   string|   null|

  139.     |   _hoodie_file_name|   string|   null|

  140.     |               rowId|   string|   null|

  141.     |         partitionId|   string|   null|

  142.     |             preComb|   bigint|   null|

  143.     |                name|   string|   null|

  144.     |           versionId|   string|   null|

  145.     |           intToLong|   bigint|   null|

  146.     |            newField|   string|   null|

  147.     +--------------------+---------+-------+

  148. 

  149. 

  150. scala> spark.sql("select rowId, partitionId, preComb, name, versionId, intToLong, newField from hudi_trips_snapshot").show()

  151.     +-----+-----------+-------+-------+---------+---------+----------+

  152.     |rowId|partitionId|preComb|   name|versionId|intToLong|  newField|

  153.     +-----+-----------+-------+-------+---------+---------+----------+

  154.     |row_3|     part_0|      0|    tom|      v_0|        0|      null|

  155.     |row_2|     part_0|      5|   john|      v_3|        3|newField_1|

  156.     |row_1|     part_0|      0|    bob|      v_0|        0|      null|

  157.     |row_5|     part_0|      5| maroon|      v_2|        2|newField_1|

  158.     |row_9|     part_0|      5|michael|      v_2|        2|newField_1|

  159.     +-----+-----------+-------+-------+---------+---------+----------+
复制代码



最新经典文章，欢迎关注公众号


链接：
https://mp.weixin.qq.com/s/4ThnSLhp2N8No2_PzaBa9A


加微信w3aboutyun，可拉入技术爱好者群
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
nettman

超级版主

关注

  • 2911

    主题

  • 3204

    帖子

  • 478

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000