在构建本地数据中心的时候，出于Apache Kudu良好的性能和兼备OLTP和OLAP的特性，以及对Impala SQL和Spark的支持，很多用户会选择Impala / Spark + Kudu的技术栈。但是由于Kudu对本地存储的依赖，导致无法支持的数据高可用和弹性扩缩容，以及社区的逐渐不活跃，越来越多的用户，开始迁移到云上的Trino / Spark + Hudi 技术栈，本文通过一个实际的例子，来看一下迁移过程中发生的代码的重构和数据的迁移。

1、现状

以下案例总结于真实场景：假设我们在帮助一个典型的零售领域的数字营销软件ISV (简称 C公司) 进行迁移，他们的数字营销软件 (平台) 借助互联网大数据、人工智能等技术，帮助他们的企业用户构建贴近客户真实行为的画像洞察。通过营销自动化精准触达和交互，提升客户体验和实现业绩增长。

大部分公司在自建数据中心的时候，会采用Cloudera Distributed Hadoop (CDH) 作为数据开发的平台，它包含常用的技术栈例如Spark，Impala，Kudu等，具体的应用场景，可以参考后面的章节。我们先来看一下这些技术栈的特点。先约定一下文中用到的几个称谓：

• 开发者：是指AWS服务的使用者和开发人员，在本文中即C公司的开发人员

• 商家：使用C公司的营销软件的企业用户，例如 S 公司，他们使用C公司的营销软件来服务他们的客户

• 客户：使用C公司的营销软件的企业用户 (例如 S 公司) 所服务的个体客户

本文中所有代码都采用TPCDS测试数据中的inventory表来作为测试数据。

1.1. CDH 介绍

CDH是Cloudera公司发行的Hadoop版本，包括Apache Hadoop生态下的常用组件，专为满足企业需求而构建。CDH提供开箱即用的企业开发所需的服务，通过将Hadoop与十几个其他关键的开源项目集成，CDH 为企业大数据开发提供了一套完整的技术栈。

同时，Cloudera创建了一个拥有集群自动化安装、中心化管理、集群监控、报警功能的一个管理软件，即 Cloudera Manager, 极大的提高了集群管理的效率。

由于CDH对原生社区的服务进行了大量的优化，明显提升了组件的稳定性和多个组件之间的兼容性，为开发人员提供了很大的方便，使得CDH成为搭建本地数据中心的首选平台。

1.2. Apache Impala 介绍

Impala是Cloudera由C++编写的基于MPP (Massively Parallel Processing) 架构的查询引擎，由运行在CDH集群上的不同的守护进程组成，它可以使用Hive的Metastore里的Database和Table等信息。Impala可以读取Hive的表数据，也可以自己创建表，特别是可以创建数据位于Kudu的表。

Impala作为流行的SQL解析引擎，其面对即席查询 (Ad-Hoc Query) 类请求的稳定性和速度在业界得到过广泛的验证。

1.3. Apache Kudu 介绍

Kudu和Impala都是Cloudera贡献给Apache基金会的顶级项目。Kudu作为底层存储，在支持高并发低延迟KV查询的同时，还保持良好的Scan性能，该特性使得其理论上能够同时兼顾OLTP类和OLAP类查询。Impala作为查询引擎，初期主要支持HDFS，Kudu发布之后，Impala和Kudu更是做了深度集成。

从下图可以看到，Kudu设计的初衷，就是想兼顾Parquet (列存，高吞吐) 和HBase (主键，低延迟) 的双重优势。

Kudu的存储架构设计，吸取了HBase的很多经验，可以参考：. 基于Kudu的存储架构，Kudu提供了良好的Upsert功能，而不需要修改整个分区的数据，这是很多开发者喜欢的，例如在数仓建模的场景中，可以频繁地修改一个Partition里的少量数据，而不是把整个Partition都Overwrite.

同时Kudu也有一些限制，例如主键的限制、分片的限制、表大小的限制……这些可以参考： . 尤其是Kudu要依赖本地的存储，不能支持HDFS或者对象存储 (例如S3) 这些高可用的存储方式，导致了Kudu在容灾备份方面考虑不充分，同时本地存储也无法实现真正的存算分离和弹性计算。因此，我们向客户推荐Hudi来替代Kudu作为存储服务。

1.4. Apache Hudi 介绍

Apache Hudi (发音为“hoodie”, 全称是：Hadoop Update Delete Incremental，以下简称为Hudi) ，作为新一代流式数据湖平台，得到了越来越广泛的应用。它是由Uber开源的项目，可以低延迟摄取数据保存到HDFS或者对象存储 (例如S3) 上。

Hudi充分利用了开源的列存储 (Parquet) 和行存储 (Avro) 的文件作为数据的存储格式，并在数据写入时生成索引，以提高查询的性能，具体请参考： .

与Kudu类似的功能是，Hudi也支持记录级别的插入更新(Upsert) 和删除，这使得Hudi能适应Kudu的很多场景。

Aamzon EMR从 6.0.0 版本和5.28.0版本开始，就提供了Hudi 组件。我们推荐使用Hudi替换Kudu的理由和场景包括：

• Spark + Hudi能实现Spark + Kudu的大部分场景，例如Upsert

• Hudi 可以将数据保存在对象存储 (例如S3) 上，对于实现存算分离和容灾备份有得天独厚的优势，同时又降低了存储的成本（Kudu需要使用本地SSD存储才能发挥高吞吐的优势）

• Hudi表支持常用的查询引擎，可以使用Hive, Presto, Trino访问Hudi表

对于Hudi的其他优势，例如Clustering, Metadata Index等，我们在这次迁移中没有使用到，在这里不展开讨论。

笔者也做了很多性能相关的测试，在同样的资源，Impala + Kudu的性能，无论是即席查询 (Ad-Hoc Query) 还是通过JDBC随机查询，都要比Trino + Hudi好一些，不过性能的问题，可以通过Amazon EMR的弹性扩容来提升和调节。

在性能之外，也需要考虑迁移后的组件的通配性和适用性。例如与其它常用组件的集成使用，以及开发和运维过程中使用的技术栈是否通用，即不会要求开发者做大量的重构代码，也不会偏离常用的和主流的技术栈，我们会保留客户大部分的Spark代码。

接下来我们会从如下两个场景，来帮助客户从Spark / Impala + Kudu的代码，迁移到Spark / Trino + Hudi上来。

2、“客户档案” 场景

客户档案是使用C公司数据分析平台的商家S公司的一个常用场景，主要通过收集客户的行为统计数据，整理客户的行为统计信息，实现客户分群，为客户打标签做数据准备。该场景下的数据特点：

1. 客户档案由3～5张表组成，由商家创建的，系统为客户生成统一的主键

2. 表的列数量为100个左右，可以自由的增加列

3. 商家会以不固定的频率更新表，例如第一天第一个批次更新100万条，第二天第二个批次更新200万条

已有的架构图如下：

2.1. 在Kudu里的实现

我们重点关注两部分，一是用户通过Java API实时写入Kudu的实现，该部分功能的测试代码如下：

……
try {
KuduTable kuduTable = kuduClient.openTable(tableName);
KuduSession kuduSession = kuduClient.newSession();
 kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
for(int i =0; i < dataArr.length; i++) {
String dataStr = dataArr[i];
Upsert upsert = kuduTable.newUpsert();
PartialRow row = upsert.getRow();
 String[] dataInfo = dataStr.split(",");
if(dataInfo.length == 4) {
 row.addInt("inv_item_sk", Integer.valueOf(dataInfo[0]));
 row.addInt("inv_warehouse_sk", Integer.valueOf(dataInfo[1]));
 row.addString("inv_date_sk", dataInfo[2]);
 row.addInt("inv_quantity_on_hand", Integer.valueOf(dataInfo[3]));
 }
 kuduSession.apply(upsert);
 }
 kuduSession.flush();
 kuduSession.close();
 System.out.println(" ******** KuduJavaSample -> upsert() successfule ! ");
} catch (KuduException e) {
 e.printStackTrace();
}
 ……

请注意，通过Java API写入Kudu表的时候，并不需要知道Kudu表完整的Schema. 另一部分就是通过Spark Kudu Client去批量操作Kudu表数据，该部分功能的测试代码如下：

……
try {
// Delete the table if it already exists.
if(kuduContext.tableExists(tableName)) {
 kuduContext.deleteTable(tableName)
 }
……
val write_rdd = spark.sparkContext.parallelize(write_arr)
val write_df = spark.createDataFrame(write_arr.asJava, schema_df)
 kuduContext.insertRows(write_df, tableName)
// Read from the table using an RDD.
val read_cols = Seq("inv_item_sk", "inv_warehouse_sk", "inv_date_sk", "inv_quantity_on_hand")
val rdd = kuduContext.kuduRDD(spark.sparkContext, tableName, read_cols)
 rdd.map { case Row(inv_item_sk: Int, inv_warehouse_sk: Int, inv_date_sk: String, inv_quantity_on_hand: Int) => (inv_item_sk, inv_warehouse_sk, inv_date_sk, inv_quantity_on_hand) }.
 collect().foreach(println(_))
// Upsert some rows.
val upsert_df = write_df.withColumn("inv_quantity_on_hand", col("inv_quantity_on_hand") + lit(1000))
 kuduContext.upsertRows(upsert_df, tableName)
// Read the table in SparkSql.
val read_df = spark.read.option("kudu.master", kuduMaster).
 option("kudu.table", tableName).
 format("kudu").load
 read_df.createOrReplaceTempView(tableName)
 spark.sqlContext.sql(s"select * from $tableName").show
……
} catch {
case unknown : Throwable => log.error(s"got an exception: " + unknown)
……

完整的测试代码请参考：, 请注意，这些测试代码只是为了展示基本的功能，省略去了业务处理逻辑的部分。

2.2. 在Hudi里的实现

我们将客户档案的架构设计中的Kudu替换为Hudi.

修改后的架构图如下：

涉及的代码重构的部分有三块：

1. Java API原来直接写入Kudu的，现在改成写入Kafka

2. 添加Spark Streaming读取Kafka数据并写入Hudi的部分

3. Spark 读写 Kudu的操作，改为Spark 读写 Hudi

其中1. 的代码可以参考：, 2) 和3)的代码可以参考： .

2.3. 组件对比

在客户档案的场景下，Kudu和Hudi两种组件的对比如下：

对比内容KuduHudi存储本地存储，无法实现存算分离和容灾备份可以存储在HDFS和对象存储 (例如S3)适配性映射到Impala表，供其它组件访问同步到Hive Metastore, 供其它组件访问JavaAPIKudu Master Server提供API需要借助Spark/Trino JDBC来访问Upsert通过JavaAPI执行Upsert，不需要Schema需要预先读取或者定义Schema3、“实时数仓” 场景

商户S公司要对客户的行为，包括点击访问、触达、订单等操作，建设数据仓库，挖掘数据价值，整理出营销活动需要的实时数据，实现提升转化率、精准投放和数据回溯。该场景下的数据特点：

1. 数据量较大，事实表 (Fact) 有时间维度

2. 大部分是新增数据，数据来源包括离线和实时两部分

3. 事实表 (Fact) 和维度表 (Dim) 的总数量为100个左右

架构图如下：

3.1. 在Kudu里的实现

从架构图上可以看出，对数据的操作分成两部分，Impala JDBC写入Kudu，这部分就是纯SQL语句；Java API实时写入Kudu, 这部分代码可以参考2.1章节中的例子。

3.2. 在Hudi里的实现

我们将实时数仓的架构设计中的Impala + Kudu替换为Spark + Hudi. 修改后的架构图如下：

涉及的代码重构的部分有三块：

1. JavaAPI原来直接写入Kudu的，现在改成写入Kafka

2. Spark Streaming 从Kafka 读取数据写入Hudi表

3. 数据仓库各个层之间的数据ETL，原来是用Impala SQL操作Kudu实现，现在改为Spark Streaming读写 Hudi表数据

其中1.的代码可以参考： . 第2.部分里，从Kafka读取数据写入ODS层的测试代码如下：

……
val df = spark
 .readStream
 .format("kafka")
 .option("kafka.bootstrap.servers", parmas.brokerList)
 .option("subscribe", parmas.sourceTopic)
 .option("startingOffsets", parmas.startPos)
 .option("failOnDataLoss", false)
 .load()
 .repartition(Integer.valueOf(parmas.partitionNum))
val schema = spark.read.format("hudi").load(s"${parmas.hudiBasePath}/${parmas.syncDB}/${parmas.syncTableName}").schema
val schema_bc = spark.sparkContext.broadcast(schema)
val tableType = if (parmas.hudiPartition !=  && parmas.hudiPartition.length > 0) "MERGE_ON_READ" else "COPY_ON_WRITE"
val query = df.writeStream
 .queryName("MSK2Hudi")
 .foreachBatch { (batchDF: DataFrame, _: Long) =>
if(batchDF !=  && (!batchDF.isEmpty) ){
val df = batchDF.withColumn("json", col("value").cast(StringType))
 .select(from_json(col("json"), schema_bc.value) as "data")
 .select("data.*")
 .withColumn("created_ts", lit((new Date()).getTime))
 .filter(genPrimaryKeyFilter(parmas.hudiKeyField))
 writeHudiTable(df, parmas.syncDB, parmas.syncTableName, "upsert", parmas.zookeeperUrl,
 parmas.hudiKeyField, "created_ts", parmas.hudiPartition, parmas.hudiBasePath, tableType)
 }
 }
 .option("checkpointLocation", parmas.checkpointDir)
 .trigger(Trigger.ProcessingTime(parmas.trigger + " seconds"))
 .start
query.awaitTermination()
……

第3.部分，从ODS层读取数据写入DW层的测试代码如下：

……
//use item as dimension table
spark.read.format("hudi").
 load("s3://dalei-demo/hudi/kudu_migration/item").
 createOrReplaceTempView("item")
while(true){
 Thread.sleep(parmas.hudiIntervel)
 endTime = DATE_FORMAT.format(new Date())
 spark.read.format("hudi").
 option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
 option(BEGIN_INSTANTTIME.key(), beginTime).
 option(END_INSTANTTIME.key(), endTime).
 load("s3://dalei-demo/hudi/kudu_migration/inventory").
 createOrReplaceTempView("inventory")
val df = spark.sql(
 s"""select in.inv_item_sk, nvl(i.i_brand, 'N/A') as i_brand, in.inv_warehouse_sk, in.inv_date_sk,
 |nvl(in.inv_quantity_on_hand, 0) as inv_quantity_on_hand,
 |${(new Date()).getTime} as created_ts
 |from inventory in left join item i on in.inv_item_sk = i.i_item_sk """.stripMargin)
if(df.count > 0)
 writeHudiTable(df, parmas.syncDB, parmas.syncTableName, "upsert", parmas.zookeeperUrl,
"inv_item_sk,i_brand,inv_warehouse_sk", "created_ts", "inv_date_sk", parmas.hudiBasePath, "MERGE_ON_READ")
 beginTime = endTime
}
……

完整的测试代码请参考： , 请注意，这些测试代码只是为了展示基本的功能，省略去了业务处理逻辑的部分。

3.3. 组件对比

在实时数仓的场景下，Kudu和Hudi两种组件的对比如下：

对比内容KuduHudi存储本地存储，无法实现存算分离和容灾备份可以存储在HDFS和对象存储 (例如S3)弹性计算无可以通过Auto Scaling 实现开发便捷Impala SQL开发比较简单Spark Dataframe 需要编程基础增量查询无，需要使用SQL从全量数据中过滤提供基于Instant Time的增量查询随机读写可以把Kudu看作一个数据库，通过Java API查询即时写入的数据需要借助Spark/Trino JDBC来实现随机读写4、数据迁移

前面章节介绍了从Kudu到Hudi的相关代码的改造，同时还需要把现存的位于Kudu的数据，迁移到Hudi上来。我们将根据不同的数据表类型，数据的量级，为客户推荐不同的迁移方案。

4.1. 迁移方案

首先，根据表的类型，选择不同的迁移方式：

• 事实表(Fact) : 初始数据的批量迁移，并通过写入Kafka的方式，实现增量数据迁移

• 维度表(Dim) : 数据变化不大，可以一次性全量迁移

• 聚合表(Aggregation) : 通过事实表和维度表计算得来，可以不用迁移，采用在目标数据库中重新计算的方式获取

其次，根据数据的量级，对于初始数据的批量迁移，可以选择不同的迁移方式：

数据量专线Snowball采用Spark 直接读写采用Kudu Export + Spark 读写< 1 TB推荐

Spark读取Kudu表数据，写入 Hudi表

< 1 PB推荐推荐Spark读取Kudu表数据，写入 Hudi表Kudu把数据导出到Parquet文件, 迁移到S3上，使用Spark写入Hudi表> 1 PB

推荐

Kudu把数据导出到Parquet文件, 迁移到S3上，使用Spark写入Hudi表

实现数据迁移的流程图如下：

4.2. 具体例子

我们来看一个实际的例子，把Kudu里的TPCDS测试数据的24张表，迁移到位于S3上Hudi表里。由于测试数据的量级是100G，所以我们采用从EMR Spark直接读取Kudu表，并写入Hudi表的方式来迁移数据。整个迁移过程耗时2小时以内。迁移的数据源和目标数据库的环境如下：

环境数据源目标数据库组件版本Kudu 1.10.0Hudi 0.10.0 (通过代码中 –packages 指定)平台CDH 6.3.2EMR 5.35.0其它组件Impala 3.2.0, Spark 2.4.5Presto 0.267, Spark 2.4.8硬件资源8 nodes m5.2xlarge8 core nodes, m6g.2xlarge

在确定CDH和EMR之间的网络连通后，开始执行迁移，具体步骤包括：

1. 初始数据的批量迁移，使用EMR 中Spark读取CDH 平台上的Kudu表，写入Hudi表

2. 将Kudu表的增量数据写入Kafka, 使用 EMR中Spark读取Kafka数据，写入Hudi表

3. 对聚合表启动实时计算

4. 迁移完成后，对比数据源和目标数据库中的数据量

步骤1.中初始化迁移的代码片段如下：

……
// get all tables in the database to migrate.
val allTable = queryByJdbc(parmas.impalaJdbcUrl + parmas.kuduDatabase, "show tables")
if(allTable !=  && allTable.isInstanceOf[Seq[String]] && allTable.length > 0) {
 allTable.foreach( tableName => {
 val (primaryKey, partitionKey) = getPrimaryAndPartitionKey(parmas.impalaJdbcUrl + parmas.kuduDatabase, tableName)
val df = spark.read
 .option("kudu.master", parmas.kuduMaster)
 .option("kudu.table", "impala::" + parmas.kuduDatabase + "." + tableName)
 .format("kudu").load
 .filter(genPrimaryKeyFilter(primaryKey))
 .withColumn("created_ts", lit((new Date()).getTime))
 .repartition(parmas.partitionNum)
val tableType = if (partitionKey !=  && partitionKey.length > 0) "MERGE_ON_READ" else "COPY_ON_WRITE"
 writeHudiTable(df, parmas.syncDB, tableName, "bulk_insert", parmas.zookeeperUrl,
 primaryKey, "created_ts", partitionKey, parmas.hudiBasePath, tableType)
 })
……

在代码中，先把指定Kudu数据库里的表选出来，然后根据Kudu表的定义，来生成Hudi表的Schema, 包括主键和分区键。这里简单的把带分区的表看作Mor表，不带分区的表看作Cow表，读者可以自己添加更加复杂的逻辑。在确定了Hudi表的类型、Schema后，调用包函数把数据写入Hudi表。步骤2. 中把Kudu表的增量数据写入Kafka的代码片段如下：

……
val df = spark.read
 .option("kudu.master", parmas.kuduMaster)
 .option("kudu.table", "impala::" + parmas.kuduDatabase + "." + parmas.syncTableName)
 .format("kudu").load
 .withColumn("created_ts", lit((new Date()).getTime))
 .filter(parmas.filterString)
 .select(to_json(struct("*"), Map("dropFieldIfAll" -> "false")).as("value"))
 .selectExpr(s"cast('${UUID.randomUUID().toString}' as string)", "cast(value as string)")
df.write
 .format("kafka")
 .option("kafka.bootstrap.servers", parmas.brokerList)
 .option("topic", parmas.sourceTopic)
 .save()
……

开发者可以通过params.filterString 来筛选增量数据。步骤2. 中从Kafka读取增量数据写入Hudi的代码片段如下：

……
val df = spark
 .readStream
 .format("kafka")
 .option("kafka.bootstrap.servers", parmas.brokerList)
 .option("subscribePattern", parmas.sourceTopic)
 .option("startingOffsets", parmas.startPos)
 .option("failOnDataLoss", false)
 .load()
 .repartition(parmas.partitionNum)
val query = df.writeStream
 .queryName("MSK2Hudi")
 .foreachBatch { (batchDF: DataFrame, _: Long) =>
if(batchDF !=  && (!batchDF.isEmpty) )
 writeMultiTable2HudiFromDF(batchDF, parmas.syncDB, "upsert", parmas.zookeeperUrl,
 parmas.hudiBasePath, parmas.impalaJdbcUrl, parmas.kuduDatabase)
 }
 .option("checkpointLocation", parmas.checkpointDir)
 .trigger(Trigger.ProcessingTime(parmas.trigger + " seconds"))
 .start
……

上面的代码中，对Kafka的Topic采用了匹配模式，可以一次读取多个Kudu表的增量数据。完整的测试代码请参考： .

4.3. 问题总结4.3.1. 版本的问题

Spark 3.x 不能读取CDH 6.3.2 上 Kudu 1.10.0的数据，所以使用EMR 5.35.0来读取，写入Hudi的时候可以通过spark-submit命令的–packages选项来指定Hudi版本为0.10.

4.3.2. 执行错误：org.apache.hudi.exception.HoodieException: (Part -) field not found in record. Acceptable fields were…….

这是因为从Kudu读出的数据，不包含precombine key导致的，可以在代码中添加一个字段作为precombine key, 值可以取当前的时间。

4.3.3. 执行错误：Caused by: java.lang.IllegalArgumentException: Can not create a Path from an empty string…….

如果Kudu没有使用Partition, 这个错会出现在Spark 2.4.8 (EMR 5.35.0) 中。考虑到没有使用Partition的表都比较小，所以全量写入Kafka, 然后从Spark 3.1.2 (EMR 6.5.0) 中读取Kafka 并写入Hudi.

4.3.4. 执行错误：To_json does not include “” value field

由于写入Kafka的数据 (value字段是json格式) 没有包含值的字段，所以跟Hudi表的Schema对不齐，可以在从Kudu表写入Kafka的时候，指定包含值的字段。也可以先从已有的Hudi表读取Schema来解析Kafka的数据，然后再写入Hudi表。

5、对于使用Kudu和Hudi的一些建议5.1. 哪些场景适合使用Kudu？

适合Kudu的场景包括：

1. 同时提供OLTP和OLAP 查询的场景

2. 高并发随机查询，尤其是查询即时写入的数据的场景

3. 用来做热数据的场景，不需要考虑数据丢失的问题

4. 简单、自用的数据仓库开发

5. 重度依赖SQL的业务人员的常用场景

5.2. 哪些场景适合使用Hudi？

适合Hudi的场景包括：

1. 注重实现弹性计算和数据安全的场景

2. 智能湖仓的场景

3. 大量使用增量查询的场景，例如较复杂的实时数仓

4. 将数据保存在对象存储 (例如S3) 上，实现多个服务组件之间数据共享的场景

5. 使用主流开源技术栈的开发场景

5.3. 可以在EMR上直接部署Kudu吗？

可以在EMR上直接部署社区版本的Impala和Kudu, 但是不推荐这样做，这样不但增加了运维的工作，还会影响EMR节点的自动扩缩容。

5.4. EMR上使用Hudi的版本

EMR上提供的Hudi依赖的jar包，其版本可以参考 , 通常来说，EMR上支持的Hudi版本会比社区稍晚一点，很多开发者喜欢在EMR使用社区的Hudi版本，这在EMR 6.5.0 以前是没有问题的。之后的EMR版本，修改了Spark操作PartitionedFile类的接口，导致与社区版本的Hudi不兼容，所以还是推荐使用EMR自带的Hudi依赖Jar包，而不是通过–packages来指定社区版本Hudi.

5.5. 总结一下迁移的好处

总的来说，从CDH上的Kudu迁移到EMR上的Hudi, 有如下好处：

1. 云原生：存算分离，弹性扩缩，成本优化

2. 稳定性：托管服务，高可用性，减少运维

3. 开放性：社区开源组件

参考文档：