利用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

1
2
3
4
5
6
7
8
9
10
11
12
13
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只有两个实现类

1
2
3
4
5
6
7
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def runMetadataBootstrap(srcPath: String, basePath: String, partitionField: String = "", extraOpts: Map[String, String] = Map.empty): Unit = {
  val bootstrapDF = spark.emptyDataFrame
  bootstrapDF.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.emptyDataFrame
  bootstrapDF.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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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)
}
1
2
3
4
5
6
7
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+---+----+-----+----+----------+
|_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文件看一下内容

1
spark.read.parquet("/tmp/bootstrap/metadata/base/f17f9a8d-a409-4023-a8a4-90a5ed21f745_1499-14-1511_00000000000001.parquet").show(false)
1
2
3
4
5
+-------------------+---------------------+------------------+----------------------+------------------------------------------------------------------------+
|_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文件

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

然后会抛出异常

1
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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)
}
1
2
3
4
5
6
7
+-------------------+---------------------+------------------+----------------------+--------------------------------------------------------------------------+---+----+-----+----+----------+
|_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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
1
2
3
4
5
6
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,还有其他配置项可以修改类型，可以自己验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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
1
2
3
4
5
6
7
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

1
2
3
4
5
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，但是博客上的示例多少还是会误导读者的~

具体看源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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);
}

@Override
public 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表的方法

可以自己写代码

1
2
3
4
5
6
7
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表时也多了一个选择。

赏

感谢您的支持!

支付宝
微信

本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

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

本文链接：https://dongkelun.com/2022/10/12/hudiBootstrap/

欢迎关注我的公众号

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~