Apache XTable 0.4 集成使用指南

目标：以 Apache Spark 为载体，给出 Apache XTable 0.4.0-incubating 的端到端集成流程——准备三种格式的源表（Hudi / Iceberg / Parquet）、执行同步、用 Spark 把同步出的目标表逐一读回验证、演示增量同步，并明确每种格式读端所需的依赖与配置。本文所有命令与回显均在 Ubuntu 22.04 + OpenJDK 17.0.20.1 + Spark 3.4.4 环境实测取得；XTable 安装见《Ubuntu 22.04 安装 Apache XTable 0.4 指南》，CLI 全参数见《Apache XTable 0.4 常用 CLI 命令与工具详解》。

一、环境信息

操作系统：Ubuntu 22.04（x86_64）
Java：OpenJDK 17.0.20.1
Spark：3.4.4（spark-submit --version 实测回显 version 3.4.4、Using Scala version 2.12.17），SPARK_HOME=/opt/spark
XTable：xtable-spark-runtime_2.12-0.4.0-incubating.jar（/opt/xtable/）
数据目录规划：/opt/data/hudi-dataset/（Hudi 源）、/opt/data/iceberg-dataset/（Iceberg 源）、/opt/data/parquet-dataset/（Parquet 源）

二、依赖版本矩阵（实测组合）

同步命令通过 --packages 引入各格式引擎库，版本必须与 XTable 0.4.0 构建矩阵对齐（实测偏差组合的报错见第八节）：

依赖	坐标	说明
Hudi 引擎包	`org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0`	HUDI 源/目标必需；不能换 1.x（0.4.0 不兼容）
Hudi Java 客户端	`org.apache.hudi:hudi-java-client:0.14.0`	HUDI 作目标时必需，缺失会被跳过
Iceberg 核心	`org.apache.iceberg:iceberg-core:1.9.2`	ICEBERG 源/目标必需；同步场景用 core 而非 spark-runtime 包
Delta	`io.delta:delta-core_2.12:2.4.0`	DELTA 源/目标必需（Spark 3.4 路径；3.5+ 走 Delta Kernel）
Avro（系统级）	`org.apache.avro:avro:1.12.0`	替换 `$SPARK_HOME/jars/avro-1.11.1.jar`（Iceberg 1.9.2 依赖 1.12 API）

三、准备源表

3.1 准备 Hudi 源表（city 分区）

脚本 /opt/scripts/create_hudi.py：

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName("create_hudi_people").getOrCreate()

table_name = "people"
local_base_path = "file:/opt/data/hudi-dataset"

records = [
   (1, 'John', 25, 'NYC', '2023-09-28 00:00:00'),
   (2, 'Emily', 30, 'SFO', '2023-09-28 00:00:00'),
   (3, 'Michael', 35, 'ORD', '2023-09-28 00:00:00'),
   (4, 'Andrew', 40, 'NYC', '2023-10-28 00:00:00'),
   (5, 'Bob', 28, 'SEA', '2023-09-23 00:00:00'),
   (6, 'Charlie', 31, 'DFW', '2023-08-29 00:00:00')
]

schema = StructType([
 StructField("id", IntegerType(), True),
 StructField("name", StringType(), True),
 StructField("age", IntegerType(), True),
 StructField("city", StringType(), True),
 StructField("create_ts", StringType(), True)
])

df = spark.createDataFrame(records, schema)

hudi_options = {
   'hoodie.table.name': table_name,
   'hoodie.datasource.write.partitionpath.field': 'city',
   'hoodie.datasource.write.hive_style_partitioning': 'true'
}

(
 df.write
   .format("hudi")
   .options(**hudi_options)
   .mode("overwrite")
   .save(f"{local_base_path}/{table_name}")
)

df2 = spark.read.format("hudi").load(f"{local_base_path}/{table_name}")
df2.select("id", "name", "age", "city", "create_ts").show()
print("ROW_COUNT:", df2.count())

spark.stop()

执行：

$SPARK_HOME/bin/spark-submit \
  --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  /opt/scripts/create_hudi.py

实测回显（节选）：

+---+-------+---+----+-------------------+
| id|   name|age|city|          create_ts|
+---+-------+---+----+-------------------+
|  6|Charlie| 31| DFW|2023-08-29 00:00:00|
|  3|Michael| 35| ORD|2023-09-28 00:00:00|
|  4| Andrew| 40| NYC|2023-10-28 00:00:00|
|  2|  Emily| 30| SFO|2023-09-28 00:00:00|
|  1|   John| 25| NYC|2023-09-28 00:00:00|
|  5|    Bob| 28| SEA|2023-09-23 00:00:00|
+---+-------+---+----+-------------------+

ROW_COUNT: 6

3.2 准备 Iceberg 源表（Hadoop 目录表）

Iceberg 源表用 Hadoop 目录型 catalog 创建，数据落在 <warehouse>/db/people/{data,metadata}（该布局与 XTable 的 --datapath 参数一一对应）。脚本 /opt/scripts/create_iceberg.py：

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName("create_iceberg_people").getOrCreate()

records = [
   (1, 'John', 25, 'NYC', '2023-09-28 00:00:00'),
   (2, 'Emily', 30, 'SFO', '2023-09-28 00:00:00'),
   (3, 'Michael', 35, 'ORD', '2023-09-28 00:00:00'),
   (4, 'Andrew', 40, 'NYC', '2023-10-28 00:00:00'),
   (5, 'Bob', 28, 'SEA', '2023-09-23 00:00:00'),
   (6, 'Charlie', 31, 'DFW', '2023-08-29 00:00:00')
]

schema = StructType([
 StructField("id", IntegerType(), True),
 StructField("name", StringType(), True),
 StructField("age", IntegerType(), True),
 StructField("city", StringType(), True),
 StructField("create_ts", StringType(), True)
])

df = spark.createDataFrame(records, schema)

(
 df.write
   .format("iceberg")
   .partitionBy("city")
   .saveAsTable("local.db.people")
)

df2 = spark.table("local.db.people")
df2.select("id", "name", "age", "city", "create_ts").orderBy("id").show()
print("ROW_COUNT:", df2.count())

spark.stop()

执行：

$SPARK_HOME/bin/spark-submit \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.9.2 \
  --conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.local.type=hadoop \
  --conf spark.sql.catalog.local.warehouse=/opt/data/iceberg-dataset \
  /opt/scripts/create_iceberg.py

实测回显：ROW_COUNT: 6；目录结构（ls /opt/data/iceberg-dataset/db/people/）：

data
metadata

⚠️ 实测排障：若直接用 df.write.format("iceberg").save(path) 按路径写，Spark 会把路径解析为命名空间并回退到 HiveCatalog，报 NoSuchNamespaceException: Namespace does not exist: file:/opt/data/iceberg-dataset（HiveCatalog 尝试访问嵌入式元数据库还会连带抛 Derby 的 MissingTableException: Required table missing : "DBS"）。用上面显式 Hadoop catalog + saveAsTable 的方式可稳定创建目录型 Iceberg 表。

3.3 准备 Parquet 源表（非空 schema）

Parquet 只读源建议用非空字段 schema 写出（全 nullable 列的 footer 会导致 XTable 列统计生成失败，实测报错见《Apache XTable 0.4 常用 CLI 命令与工具详解》第六节）。脚本 /opt/scripts/create_items.py：

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName("create_items_nonnull").getOrCreate()

items_schema = StructType([
 StructField("item_id", IntegerType(), False),
 StructField("item_name", StringType(), False),
 StructField("price", DoubleType(), False)
])
items = [
 (1, 'keyboard', 199.0),
 (2, 'mouse', 89.0),
 (3, 'monitor', 1299.0)
]
df2 = spark.createDataFrame(items, items_schema)
df2.write.mode("overwrite").parquet("file:/opt/data/parquet-dataset/items2")
n2 = spark.read.parquet("file:/opt/data/parquet-dataset/items2").count()
print("ITEMS2_ROW_COUNT:", n2)

spark.stop()

执行：$SPARK_HOME/bin/spark-submit /opt/scripts/create_items.py，实测回显 ITEMS2_ROW_COUNT: 3。

四、执行同步

三张源表的同步命令（--packages 组合见第二节）：

# Hudi 源（分区表需 --partitionspec）
$SPARK_HOME/bin/spark-submit \
  --class org.apache.xtable.spark.XTableSparkSync \
  --master 'local[1]' \
  --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0,org.apache.iceberg:iceberg-core:1.9.2,io.delta:delta-core_2.12:2.4.0 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/xtable/xtable-spark-runtime_2.12-0.4.0-incubating.jar \
  --basepath /opt/data/hudi-dataset/people \
  --sourceformat HUDI \
  --targets ICEBERG,DELTA \
  --partitionspec city:VALUE

# Iceberg 源（数据在 data 子目录，需 --datapath；HUDI 目标需 hudi-java-client）
$SPARK_HOME/bin/spark-submit \
  --class org.apache.xtable.spark.XTableSparkSync \
  --master 'local[1]' \
  --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0,org.apache.hudi:hudi-java-client:0.14.0,org.apache.iceberg:iceberg-core:1.9.2,io.delta:delta-core_2.12:2.4.0 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/xtable/xtable-spark-runtime_2.12-0.4.0-incubating.jar \
  --basepath /opt/data/iceberg-dataset/db/people \
  --datapath /opt/data/iceberg-dataset/db/people/data \
  --sourceformat ICEBERG \
  --targets HUDI,DELTA

# Parquet 源
$SPARK_HOME/bin/spark-submit \
  --class org.apache.xtable.spark.XTableSparkSync \
  --master 'local[1]' \
  --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0,org.apache.hudi:hudi-java-client:0.14.0,org.apache.iceberg:iceberg-core:1.9.2,io.delta:delta-core_2.12:2.4.0 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/xtable/xtable-spark-runtime_2.12-0.4.0-incubating.jar \
  --basepath /opt/data/parquet-dataset/items2 \
  --sourceformat PARQUET \
  --targets ICEBERG,DELTA,HUDI

三条命令的实测成功回显分别为：

INFO ConversionController: Sync is successful for the following formats ICEBERG,DELTA
INFO ConversionController: Sync is successful for the following formats HUDI,DELTA
INFO ConversionController: Sync is successful for the following formats HUDI,ICEBERG,DELTA

五、目标表读回验证

每种格式读端要求不同，实测全部通过：

5.1 读 Delta 目标（Hudi 源同步产物）

Delta 分区表读取需配置 Delta 的 catalog 扩展：

$SPARK_HOME/bin/spark-submit \
  --packages io.delta:delta-core_2.12:2.4.0 \
  --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
  /opt/scripts/read_back.py delta /opt/data/hudi-dataset/people

实测回显：

+---+-------+---+----+-------------------+
| id|   name|age|city|          create_ts|
+---+-------+---+----+-------------------+
|  1|   John| 25| NYC|2023-09-28 00:00:00|
|  2|  Emily| 30| SFO|2023-09-28 00:00:00|
|  3|Michael| 35| ORD|2023-09-28 00:00:00|
|  4| Andrew| 40| NYC|2023-10-28 00:00:00|
|  5|    Bob| 28| SEA|2023-09-23 00:00:00|
|  6|Charlie| 31| DFW|2023-08-29 00:00:00|
+---+-------+---+----+-------------------+

DELTA_ROW_COUNT: 6

（读端脚本逻辑：spark.read.format(fmt).load(path) 后 show() 与 count()。）

5.2 读 Iceberg 目标

Spark 读 Iceberg 需要 iceberg-spark-runtime（与同步用的 iceberg-core 不同，读端是 SQL 层接入）：

$SPARK_HOME/bin/spark-submit \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.9.2 \
  /opt/scripts/read_back.py iceberg /opt/data/hudi-dataset/people

实测回显与 Delta 一致：6 行数据（id 1~6），ICEBERG_ROW_COUNT: 6。

5.3 读 Hudi 目标（Iceberg 源同步产物）

关键配置：读 XTable 同步出的 Hudi 表必须显式开启元数据表选项，否则实测结果为空（TableNotFoundException 或 0 行）：

$SPARK_HOME/bin/spark-submit \
  --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/scripts/read_back.py hudi /opt/data/iceberg-dataset/db/people/data

读端脚本中 Hudi 分支的选项（实测必须）：

df = (
    spark.read.format("hudi")
    .option("hoodie.metadata.enable", "true")
    .option("hoodie.datasource.write.hive_style_partitioning", "true")
    .load(path)
)

实测回显：

+---+-------+---+----+-------------------+
| id|   name|age|city|          create_ts|
+---+-------+---+----+-------------------+
|  1|   John| 25| NYC|2023-09-28 00:00:00|
|  2|  Emily| 30| SFO|2023-09-28 00:00:00|
|  3|Michael| 35| ORD|2023-09-28 00:00:00|
|  4| Andrew| 40| NYC|2023-10-28 00:00:00|
|  5|    Bob| 28| SEA|2023-09-23 00:00:00|
|  6|Charlie| 31| DFW|2023-08-29 00:00:00|
+---+-------+---+----+-------------------+

HUDI_ROW_COUNT: 6

5.4 Parquet 源产物的读回结果

实测 Parquet 源同步出的三个目标：Hudi 目标读回 3 行正常（HUDI_ROW_COUNT: 3）；Delta/Iceberg 目标因 PARQUET 源未填充文件行数统计（numRecords 为 0），count() 返回 0（详见《Apache XTable 0.4 常用 CLI 命令与工具详解》第六节与本文第八节）。

六、增量同步实测

XTable 支持增量同步：源表新增提交后再次执行同一同步命令，只处理新提交并更新目标端水位。

向 Hudi 源表追加 2 行（脚本逻辑：createDataFrame 两行新数据后 .write.format("hudi").mode("append") 写回同一路径），实测 PEOPLE_ROW_COUNT_AFTER_APPEND: 8。
重跑第四节第一条 Hudi 源同步命令，实测成功回显同前；Delta 事务日志从 00000000000000000000.json 增至 00000000000000000001.json（实测 _delta_log/ 目录两个文件并存）。
新版本事务日志中记录了增量水位（实测节选）：

"tags":{"XTABLE_METADATA":"{\"lastInstantSynced\":\"2026-10-15T02:58:32.592Z\",\"instantsToConsiderForNextSync\":[],\"version\":0,\"sourceTableFormat\":\"HUDI\",\"sourceIdentifier\":\"20261015025832592\"}"}

读回验证两个目标均更新为 8 行：实测 DELTA_ROW_COUNT: 8、ICEBERG_ROW_COUNT: 8（首次同步后为 6 行）。

七、同步产物的元数据解读

Hudi 源表目录（/opt/data/hudi-dataset/people/）：.hoodie/（源时间线）、city=*/（数据文件）、metadata/（Iceberg 目标）、_delta_log/（Delta 目标）。
Iceberg 源表（--datapath 模式）：目标元数据写在 data/ 下，即 /opt/data/iceberg-dataset/db/people/data/{.hoodie,_delta_log}；读取目标表也用该路径。
Hudi 目标的提交记录：外部数据文件的 path 字段以 _hudiext 后缀登记（后缀前为提交时刻时间戳），这是 Hudi 0.14 的外部文件机制，读端无需额外处理（保持 0.14.0 + hoodie.metadata.enable=true 即可）。
Delta 目标的提交记录：operation 为 UPDATE，operationParameters.predicate 为 ["xtable-delta-sync"]，engineInfo 为 Apache-Spark/3.4.4 Delta-Lake/2.4.0，可用于日志审计。
Hudi 目标的 hoodie.properties（实测节选）：hoodie.table.type=COPY_ON_WRITE、hoodie.table.partition.fields=city、hoodie.table.timeline.layout.version=1、hoodie.datasource.write.hive_style_partitioning=true。

八、限制与排障（集成场景实测）

引擎版本必须对齐矩阵：Hudi 1.2.0 的包会报 NoSuchMethodError: HoodieTableMetaClient$Builder.setConf；Hudi 1.x 写出的源表报 IllegalArgumentException（TimelineLayoutVersion）；Iceberg 源配 iceberg-spark-runtime 包报 ClassCastException: PartitionData cannot be cast to IndexedRecord（改用 iceberg-core）；Spark 自带 Avro 1.11 报 NoSuchMethodError: LogicalTypes$TimestampNanos。逐项解法见《Ubuntu 22.04 安装 Apache XTable 0.4 指南》与《Apache XTable 0.4 常用 CLI 命令与工具详解》的常见问题节。
只支持 COW / Read-Optimized 视图：MOR 表的 log 文件、Delta/Iceberg 的 deletion vectors 不同步。
读 Hudi 目标三要素：Hudi 0.14.0+、hoodie.metadata.enable=true、hive 风格分区，缺一实测结果为空。
PARQUET 源的行数统计：Delta/Iceberg 目标 numRecords=0，count() 失真；Hudi 目标正常。

九、参考链接

Spark Runtime 使用指南：https://xtable.apache.org/docs/how-to-spark-runtime
Quick Start（三格式建表示例）：https://xtable.apache.org/docs/how-to
从 Spark 查询同步表：https://xtable.apache.org/docs/spark
功能与限制：https://xtable.apache.org/docs/features-and-limitations
0.4.0 发布说明：https://xtable.apache.org/releases/release-0.4.0-incubating
源码仓库：https://github.com/apache/incubator-xtable