利用Hudi Bootstrap转化现有Hive表的parquet/orc文件为Hudi表

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站：https://www.captainai.net/dongkelun

前言

在最开始学习Hudi源码时，就发现了Hudi有一个Bootstrap功能，但是一直没用过，通过官网文档https://hudi.apache.org/cn/docs/migration_guide/可知,它可以将现有的表件转化为Hudi表，而且有两种类型METADATA_ONLY和FULL_RECORD，但是文档并不详细，比如这两种类型的区别具体是啥，支持哪些文件类型的源表。于是带着这些疑问来学习一下它是如何使用的以及源码原理的实现，这样可以更全面的了解Hudi。

版本

Hudi 0.12.0
Spark 2.4.4/3.1.2

支持的文件类型

如题，目前只支持两种文件类型：parquet和orc,对于其他格式的Hive表，比如text就不能用Bootstrap进行转化了
其实官方文档并没有说明支持哪些类型，我们可以在源码里找到答案：

METADATA_ONLY

  public static BootstrapMetadataHandler getMetadataHandler(HoodieWriteConfig config, HoodieTable table, HoodieFileStatus srcFileStatus) {Path sourceFilePath = FileStatusUtils.toPath(srcFileStatus.getPath());String extension = FSUtils.getFileExtension(sourceFilePath.toString());BootstrapMetadataHandler bootstrapMetadataHandler;if (ORC.getFileExtension().equals(extension)) {return new OrcBootstrapMetadataHandler(config, table, srcFileStatus);} else if (PARQUET.getFileExtension().equals(extension)) {return new ParquetBootstrapMetadataHandler(config, table, srcFileStatus);} else {throw new HoodieIOException("Bootstrap Metadata Handler not implemented for base file format " + extension);}}

FULL_RECORD

FullRecordBootstrapDataProvider只有两个实现类

public abstract class FullRecordBootstrapDataProvider<I> implements Serializable {public abstract class SparkFullBootstrapDataProviderBase extends FullRecordBootstrapDataProvider<JavaRDD<HoodieRecord>> {// 对应paruqet文件类型
public class SparkParquetBootstrapDataProvider extends SparkFullBootstrapDataProviderBase {// 对应orc文件类型
public class SparkOrcBootstrapDataProvider extends SparkFullBootstrapDataProviderBase {

Bootstrap 类型

METADATA_ONLY：只会生成主键、页脚的基本框架文件，不会重写全部数据，这样可以节省时间，提升效率，而查询时，会根据元数据信息去查源表的文件，也就是原来的表数据文件不能删除，这算是它的一个弊端，不过当有后续数据写入，第一次commit对应的parquet文件变为历史文件时，删除源表的文件就不会影响正常读取了。_hoodie_commit_time为00000000000001,对应源码HoodieTimeline.METADATA_BOOTSTRAP_INSTANT_TS
FULL_RECORD：首先读取源表文件生成RDD，然后将RDD以bulkInsert的形式写到新的Hudi表，也就是以Hudi表的形式执行数据的完整复制/重写。_hoodie_commit_time为00000000000002,对应源码HoodieTimeline.FULL_BOOTSTRAP_INSTANT_TS

不管用哪种类型，我们后续都可以在这个转化的Hudi表上执行upsert等各种写操作和查询，需要注意的是要求源表不能有重复数据，因为利用Hudi Bootstrap转化Hudi表时不会去重，否则我们在后续的upsert时，也会有重复数据。

使用方式一：Spark代码

完整代码已提交到github：https://github.com/dongkelun/hudi-demo

  def runMetadataBootstrap(srcPath: String, basePath: String, partitionField: String = "", extraOpts: Map[String, String] = Map.empty): Unit = {val bootstrapDF = spark.emptyDataFramebootstrapDF.write.format("hudi").options(extraOpts).option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.BOOTSTRAP_OPERATION_OPT_VAL).option(HoodieWriteConfig.TBL_NAME.key, tableName).option(DataSourceWriteOptions.RECORDKEY_FIELD.key, "id").option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key, partitionField).option(HoodieBootstrapConfig.BASE_PATH.key, srcPath).option(HoodieBootstrapConfig.KEYGEN_CLASS_NAME.key, classOf[ComplexKeyGenerator].getName).option(KEYGENERATOR_CLASS_NAME.key(), classOf[ComplexKeyGenerator].getName).// 同步Hive相关的配置option(HoodieSyncConfig.META_SYNC_ENABLED.key(), true).option(HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key, true).option(HiveSyncConfigHolder.HIVE_SYNC_MODE.key, HiveSyncMode.HMS.name()).option(HoodieSyncConfig.META_SYNC_DATABASE_NAME.key, "test").option(HiveSyncConfigHolder.HIVE_AUTO_CREATE_DATABASE.key(), true).option(HiveSyncConfigHolder.HIVE_CREATE_MANAGED_TABLE.key, true).option(HoodieSyncConfig.META_SYNC_TABLE_NAME.key, tableName).option(HIVE_STYLE_PARTITIONING.key, true).option(HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key, partitionField).option(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key, classOf[MultiPartKeysValueExtractor].getName).mode(SaveMode.Overwrite).save(basePath)}def runFullBootstrap(srcPath: String, basePath: String, partitionField: String = "", extraOpts: Map[String, String] = Map.empty): Unit = {val bootstrapDF = spark.emptyDataFramebootstrapDF.write.format("hudi").options(extraOpts).option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.BOOTSTRAP_OPERATION_OPT_VAL).option(HoodieWriteConfig.TBL_NAME.key, tableName).option(DataSourceWriteOptions.RECORDKEY_FIELD.key, "id").option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key, partitionField).option(HoodieBootstrapConfig.BASE_PATH.key, srcPath).option(HoodieBootstrapConfig.KEYGEN_CLASS_NAME.key, classOf[ComplexKeyGenerator].getName).option(HoodieBootstrapConfig.MODE_SELECTOR_CLASS_NAME.key, classOf[FullRecordBootstrapModeSelector].getName).option(KEYGENERATOR_CLASS_NAME.key(), classOf[ComplexKeyGenerator].getName).// 同步Hive相关的配置option(HoodieSyncConfig.META_SYNC_ENABLED.key(), true).option(HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key, true).option(HiveSyncConfigHolder.HIVE_SYNC_MODE.key, HiveSyncMode.HMS.name()).option(HoodieSyncConfig.META_SYNC_DATABASE_NAME.key, "test").option(HiveSyncConfigHolder.HIVE_AUTO_CREATE_DATABASE.key(), true).option(HiveSyncConfigHolder.HIVE_CREATE_MANAGED_TABLE.key, true).option(HoodieSyncConfig.META_SYNC_TABLE_NAME.key, tableName).option(HIVE_STYLE_PARTITIONING.key, true).option(HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key, partitionField).option(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key, classOf[MultiPartKeysValueExtractor].getName).mode(SaveMode.Overwrite).save(basePath)}

METADATA_ONLY

  def testMetadataBootstrapCOWHiveStylePartitioned2(): Unit = {import spark.implicits._val srcPath = "/tmp/bootstrap/metadata/src_hiveStylePartition2"val basePath = "/tmp/bootstrap/metadata/base_hiveStylePartition2"val sourceDF = Seq((1, "a1", 10, 1000, "2022-10-08"), (2, "a2", 20, 2000, "2022-10-09")).toDF("id", "name", "value", "ts", "dt")val partitions = Seq("2022-10-08", "2022-10-09")partitions.foreach(partition => {sourceDF.filter(sourceDF("dt").equalTo(lit(partition))).write.format("parquet").mode(SaveMode.Overwrite).save(srcPath + "/dt=" + partition)})val extraOpts = Map(DataSourceWriteOptions.HIVE_STYLE_PARTITIONING.key -> "true")runMetadataBootstrap(srcPath, basePath, "dt", extraOpts)spark.read.format("hudi").load(basePath).show(false)val dfAppend = Seq((2, "a2", 22, 2200, "2022-10-09"), (4, "a4", 40, 4000, "2022-10-08")).toDF("id", "name", "value", "ts", "dt")append2Hudi(dfAppend, basePath, "dt", extraOpts)spark.read.format("hudi").load(basePath).show(false)}

+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|_hoodie_commit_time|_hoodie_commit_seqno |_hoodie_record_key|_hoodie_partition_path|_hoodie_file_name                                                       |id |name|value|ts  |dt        |
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|00000000000001     |00000000000001_749_0 |id:1              |dt=2022-10-08         |722dae6a-9c27-4976-b9a4-b1eef3800c45_1-48-2744_20221012103644857.parquet|1  |a1  |10   |1000|2022-10-08|
|20221012103644857  |20221012103644857_1_1|id:4              |dt=2022-10-08         |722dae6a-9c27-4976-b9a4-b1eef3800c45_1-48-2744_20221012103644857.parquet|4  |a4  |40   |4000|2022-10-08|
|20221012103644857  |20221012103644857_0_0|id:2              |dt=2022-10-09         |be5a9c32-705b-43c6-b9fd-6882532af47e_0-48-2743_20221012103644857.parquet|2  |a2  |22   |2200|2022-10-09|
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+

问题：hive分区表对应的parquet文件中没有分区字段，对于这种情况，Spark读取转化后的Hudi表时不能读取分区字段值，不过同步到Hive表，用Hive SQL 查询是可以正常查到hive分区字段的，解决方式：修改源码使Spark Hudi Bootstartp读写支持配置项hoodie.datasource.write.drop.partition.columns,参考PR：https://github.com/apache/hudi/pull/5201

验证逻辑

前提：先不要append，否则会变为历史数据

首先直接读取Hudi表下面的parquet文件看一下内容

spark.read.parquet("/tmp/bootstrap/metadata/base/f17f9a8d-a409-4023-a8a4-90a5ed21f745_1499-14-1511_00000000000001.parquet").show(false)

+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+
|_hoodie_commit_time|_hoodie_commit_seqno |_hoodie_record_key|_hoodie_partition_path|_hoodie_file_name                                                       |
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+
|00000000000001     |00000000000001_1499_0|id:1              |                      |f17f9a8d-a409-4023-a8a4-90a5ed21f745_1499-14-1511_00000000000001.parquet|
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+

确实只有元数据信息，然后再删除源表的parquet文件

spark.read.format("hudi").load("/tmp/bootstrap/metadata/base").show()

然后会抛出异常

java.io.FileNotFoundException: File file:/tmp/bootstrap/metadata/src/part-00000-5429fabc-94be-4bc0-b1d7-f15adff0a619-c000.snappy.parquet does not exist

说明读取时会读源表的parquet文件

FULL_RECORD

  def testFullBootstrapCOWHiveStylePartitioned2(): Unit = {import spark.implicits._val srcPath = "/tmp/bootstrap/full/src_hiveStylePartition2"val basePath = "/tmp/bootstrap/full/base_hiveStylePartition2"val sourceDF = makeDfColNullable(Seq((1, "a1", 10, 1000, "2022-10-08"),(2, "a2", 20, 2000, "2022-10-09")).toDF("id", "name", "value", "ts", "dt"))val partitions = Seq("2022-10-08", "2022-10-09")partitions.foreach(partition => {sourceDF.filter(sourceDF("dt").equalTo(lit(partition))).write.format("parquet").mode(SaveMode.Overwrite).save(srcPath + "/dt=" + partition)})val extraOpts = Map(DataSourceWriteOptions.HIVE_STYLE_PARTITIONING.key -> "true")runFullBootstrap(srcPath, basePath, "dt", extraOpts)spark.read.format("hudi").load(basePath).show(false)val dfAppend = Seq((2, "a2", 22, 2200, "2022-10-09"), (4, "a4", 40, 4000, "2022-10-08")).toDF("id", "name", "value", "ts", "dt")append2Hudi(dfAppend, basePath, "dt", extraOpts)spark.read.format("hudi").load(basePath).show(false)}

+-------------------+---------------------+------------------+----------------------+--------------------------------------------------------------------------+---+----+-----+----+----------+
|_hoodie_commit_time|_hoodie_commit_seqno |_hoodie_record_key|_hoodie_partition_path|_hoodie_file_name                                                         |id |name|value|ts  |dt        |
+-------------------+---------------------+------------------+----------------------+--------------------------------------------------------------------------+---+----+-----+----+----------+
|00000000000002     |00000000000002_0_0   |id:1              |dt=2022-10-08         |f367185f-072f-4c27-ab02-affe728421ed-0_1-49-1250_20221012110352241.parquet|1  |a1  |10   |1000|2022-10-08|
|20221012110352241  |20221012110352241_1_1|id:4              |dt=2022-10-08         |f367185f-072f-4c27-ab02-affe728421ed-0_1-49-1250_20221012110352241.parquet|4  |a4  |40   |4000|2022-10-08|
|20221012110352241  |20221012110352241_0_0|id:2              |dt=2022-10-09         |e5266643-a24f-46e8-aaee-e820334de9b9-0_0-49-1249_20221012110352241.parquet|2  |a2  |22   |2200|2022-10-09|
+-------------------+---------------------+------------------+----------------------+--------------------------------------------------------------------------+---+----+-----+----+----------+

问题：1、hive分区表对应的parquet文件中没有分区字段，对于这种情况FULL_RECORD目前还不支持，解决方式：修改源码使Spark Hudi Bootstartp读写支持配置项hoodie.datasource.write.drop.partition.columns,参考PR：https://github.com/apache/hudi/pull/5201
2、对于parquet文件内容有要求，当sourceDF存在不是String类型的字段，且nullable=false时，转化的Hudi表会乱码，所以示例代码里先将sourceDF的所有字段的nullable设置成了true，原因应该是因为源码里读取源表parquet文件时对某些schema不兼容，修改读取parquet文件相关的源码应该就可以解决

使用方式2：HoodieDeltaStreamer

METADATA_ONLY

bin/spark-submit \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer jars/hudi-utilities-bundle_2.12-0.12.0.jar \
--run-bootstrap \
--target-base-path /tmp/test_hudi_bootstrap_metadata \
--target-table test_hudi_bootstrap_metadata \
--table-type COPY_ON_WRITE \
--hoodie-conf hoodie.bootstrap.base.path=/tmp/bootstrap/metadata/src_partition \
--hoodie-conf hoodie.datasource.write.recordkey.field=id \
--hoodie-conf hoodie.datasource.write.partitionpath.field=dt \
--hoodie-conf hoodie.datasource.write.hive_style_partitioning=true \
--hoodie-conf hoodie.bootstrap.keygen.class=org.apache.hudi.keygen.ComplexKeyGenerator \
--enable-sync  \
--hoodie-conf hoodie.datasource.hive_sync.mode=HMS \
--hoodie-conf hoodie.datasource.hive_sync.database=test \
--hoodie-conf hoodie.datasource.hive_sync.auto_create_database=true \
--hoodie-conf hoodie.datasource.hive_sync.create_managed_table=true \
--hoodie-conf hoodie.datasource.hive_sync.table=test_hudi_bootstrap_metadata \
--hoodie-conf hoodie.datasource.hive_sync.partition_fields=dt \
--hoodie-conf hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor

select * from test_hudi_bootstrap_metadata;
+----------------------+------------------------+---------------------+-------------------------+----------------------------------------------------+-----+-------+--------+-------+-------------+
| _hoodie_commit_time  |  _hoodie_commit_seqno  | _hoodie_record_key  | _hoodie_partition_path  |                 _hoodie_file_name                  | id  | name  | value  |  ts   |     dt      |
+----------------------+------------------------+---------------------+-------------------------+----------------------------------------------------+-----+-------+--------+-------+-------------+
| 00000000000001       | 00000000000001_749_0   | 1                   | 2022-10-08              | ffccead0-2f32-4959-9687-edb7b1b8cbf3_749-14-762_00000000000001.parquet | 1   | a1    | 10     | 1000  | 2022-10-08  |
| 00000000000001       | 00000000000001_1499_0  | 2                   | 2022-10-09              | bc7f9b9f-3df0-48d5-beac-05e7d1a32fc2_1499-14-1512_00000000000001.parquet | 2   | a2    | 20     | 2000  | 2022-10-09  |

问题：这里设置的hive_style_partitioning=true没有生效

FULL_RECORD

只需要配置：hoodie.bootstrap.mode.selector=org.apache.hudi.client.bootstrap.selector.FullRecordBootstrapModeSelector，类型就会变成FULL_RECORD,还有其他配置项可以修改类型，可以自己验证

bin/spark-submit \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer jars/hudi-utilities-bundle_2.12-0.12.0.jar \
--run-bootstrap \
--target-base-path /tmp/test_hudi_bootstrap_full \
--target-table test_hudi_bootstrap_full \
--table-type COPY_ON_WRITE \
--hoodie-conf hoodie.bootstrap.mode.selector=org.apache.hudi.client.bootstrap.selector.FullRecordBootstrapModeSelector \
--hoodie-conf hoodie.bootstrap.base.path=/tmp/bootstrap/full/src_hiveStylePartition2 \
--hoodie-conf hoodie.datasource.write.recordkey.field=id \
--hoodie-conf hoodie.datasource.write.partitionpath.field=dt \
--hoodie-conf hoodie.datasource.write.hive_style_partitioning=true \
--hoodie-conf hoodie.bootstrap.keygen.class=org.apache.hudi.keygen.ComplexKeyGenerator \
--enable-sync  \
--hoodie-conf hoodie.datasource.hive_sync.mode=HMS \
--hoodie-conf hoodie.datasource.hive_sync.database=test \
--hoodie-conf hoodie.datasource.hive_sync.auto_create_database=true \
--hoodie-conf hoodie.datasource.hive_sync.create_managed_table=true \
--hoodie-conf hoodie.datasource.hive_sync.table=test_hudi_bootstrap_full \
--hoodie-conf hoodie.datasource.hive_sync.partition_fields=dt \
--hoodie-conf hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor

select * from test_hudi_bootstrap_full;
+----------------------+-----------------------+---------------------+-------------------------+----------------------------------------------------+-----+-------+--------+-------+-------------+
| _hoodie_commit_time  | _hoodie_commit_seqno  | _hoodie_record_key  | _hoodie_partition_path  |                 _hoodie_file_name                  | id  | name  | value  |  ts   |     dt      |
+----------------------+-----------------------+---------------------+-------------------------+----------------------------------------------------+-----+-------+--------+-------+-------------+
| 00000000000002       | 00000000000002_0_0    | 1                   | dt=2022-10-08           | 6df53d90-c04c-455e-b57d-32312344f701-0_0-15-14_00000000000002.parquet | 1   | a1    | 10     | 1000  | 2022-10-08  |
| 00000000000002       | 00000000000002_1_0    | 2                   | dt=2022-10-09           | fe02f500-3291-446c-bd5c-957dd60db86d-0_1-15-15_00000000000002.parquet | 2   | a2    | 20     | 2000  | 2022-10-09  |
+----------------------+-----------------------+---------------------+-------------------------+----------------------------------------------------+-----+-------+--------+-------+-------------+

官网博客上的问题

地址：https://hudi.apache.org/cn/blog/2020/08/20/efficient-migration-of-large-parquet-tables/#known-caveats

Here is an example for running METADATA_ONLY bootstrap using Delta Streamer.
--hoodie-conf hoodie.bootstrap.full.input.provider=org.apache.hudi.bootstrap.SparkParquetBootstrapDataProvider \
--hoodie-conf hoodie.bootstrap.mode.selector=org.apache.hudi.client.bootstrap.selector.BootstrapRegexModeSelector \
--hoodie-conf hoodie.bootstrap.mode.selector.regex="2020/08/2[0-9]" \
--hoodie-conf hoodie.bootstrap.mode.selector.regex.mode=METADATA_ONLY

这里说的是METADATA_ONLY的例子，经验证实际为FULL_RECORD,虽然配置了regex.mode=METADATA_ONLY，其实默认值就是METADATA_ONLY，但是不生效，原因是因为配置了hoodie.bootstrap.mode.selector=org.apache.hudi.client.bootstrap.selector.BootstrapRegexModeSelector，这个类里的defaultMode的逻辑应该有问题，我已经提交了PR：https://github.com/apache/hudi/pull/6928

更新：对于BootstrapRegexModeSelector我理解有误，它的逻辑应该是这样的：
首先有配置：hoodie.bootstrap.mode.selector.regex.mode 默认值METADATA_ONLY、hoodie.bootstrap.mode.selector.regex默认值.*
但是如果不是默认值的话，比如上面的2020/08/2[0-9],假设我们有分区"2020/08/10,2020/08/10/11,2020/08/20,2020/08/21",那么匹配成功的2020/08/20和2020/08/21对应的类型为METADATA_ONLY，匹配不成功的2020/08/10和2020/08/10/11则为FULL_RECORD。而至于我的为啥都是FULL_RECORD，原因是regex设置错误，我设置的是2022/10/0[0-9],但实际的分区值为2022-10-08和2022-10-09(分隔符不一样)，而如果用默认的.*的话，则全部能匹配上，也就都是METADATA_ONLY，但是博客上的示例多少还是会误导读者的~

具体看源码：

  public BootstrapRegexModeSelector(HoodieWriteConfig writeConfig) {super(writeConfig);this.pattern = Pattern.compile(writeConfig.getBootstrapModeSelectorRegex());this.bootstrapModeOnMatch = writeConfig.getBootstrapModeForRegexMatch();this.defaultMode = BootstrapMode.FULL_RECORD.equals(bootstrapModeOnMatch)? BootstrapMode.METADATA_ONLY : BootstrapMode.FULL_RECORD;LOG.info("Default Mode :" + defaultMode + ", on Match Mode :" + bootstrapModeOnMatch);}@Overridepublic Map<BootstrapMode, List<String>> select(List<Pair<String, List<HoodieFileStatus>>> partitions) {return partitions.stream()// 匹配上的话，值为bootstrapModeOnMatch，默认为METADATA_ONLY，否则为defaultMode，也就是另外一种类型`FULL_RECORD`// bootstrapModeOnMatch 和 defaultMode是对立的.map(p -> Pair.of(pattern.matcher(p.getKey()).matches() ? bootstrapModeOnMatch : defaultMode, p.getKey())).collect(Collectors.groupingBy(Pair::getKey, Collectors.mapping(Pair::getValue, Collectors.toList())));}

其他转化现有表为Hudi表的方法

可以自己写代码

val sourceDf = spark.table("tableName")
sourceDf.write.format("hudi").option()....saveAsTable("tableName")
// 或
for partition in [list of partitions in source table] {val inputDF = spark.read.format("any_input_format").load("partition_path")inputDF.write.format("org.apache.hudi").option()....save("basePath")
}

也可以使用HoodieDeltaStreamer的其他功能，可以参考我之前写的文章：Hudi DeltaStreamer使用总结

总结

本文介绍了如何利用Hudi Bootstrap转化现有Hive表为Hudi表，提供了完整的代码示例，并分析了METADATA_ONLY和FULL_RECORD的区别以及利弊，指出了目前版本存在的一些问题，虽然有着些许问题，但是我相信社区以后肯定会完善解决这些问题，这样我们转化Hudi表时也多了一个选择。