Migrating from Apache Hudi-based data lake to Apache Iceberg-based data lake

A cloud data lake is a scalable, low-cost place to store data from many different sources. Over the years, data lakes built on Amazon Simple Storage Service (Amazon S3) have become a common home for enterprise data, used by teams who query it for analytics and machine learning. However, data lakes have historically made it hard to update and delete records, keep transactions isolated, and handle concurrent changes safely. Open table formats such as Apache Hudi, Delta Lake, and Apache Iceberg were created to solve these problems. They add ACID transactions and row-level updates and delete, along with snapshot history and schema enforcement that make data lakes easier to audit and govern.

As organizations’ data architectures mature and evolve, the need to migrate between these table formats becomes increasingly common. This transition is often driven by changing requirements, performance considerations, or the desire to use newer features. Among these migrations, the shift from Apache Hudi to Apache Iceberg has gained particular attention, as organizations seek to capitalize on Iceberg’s modern architecture and advanced capabilities.

In this post, we show how to convert Apache Hudi tables to Apache Iceberg, using a Jupyter notebook running on AWS Glue 5.1 interactive sessions.

Benefits of migration to Iceberg

Enhanced partition evolution

Iceberg supports dynamic changes to partitioning schemes without requiring complete data rewrites. This capability is valuable in evolving data environments where query patterns change over time. You can adjust your partitioning strategy to optimize performance, reducing the operational overhead traditionally associated with such changes. Your data lake can adapt to changing business requirements without costly reorganization efforts.

Hidden partitioning

In addition to standard partitioning on an existing column, Iceberg supports hidden partitioning, which decouples the physical partition layout from the columns that queries filter on. The table specification declares a partition transform, which is a rule that tells Iceberg how to derive the partition value from a source column, such as a time bucket or a hash of the column value. The transform is defined once at the table level, and Iceberg applies it automatically on every write. This removes a long-standing source of friction on both sides of the table: writers no longer need to extract and maintain a separate partition column alongside the real one at ingestion time, and queries no longer need to carry an extra predicate on that derived column to get good performance.

Openness with the Iceberg REST Catalog

The Iceberg REST Catalog specification defines a standard HTTP protocol for catalog operations. Any engine that speaks the protocol can talk to any catalog that implements it, without a catalog-specific client library or implementation JAR on the class path. Integration moves from a code and dependency problem to a configuration problem, making Iceberg a strong fit for multi-engine data architectures.

Deep integration with AWS services

Iceberg has native integration across the AWS analytics ecosystem. Amazon Athena, Amazon Redshift, Amazon EMR, and AWS Glue all read and write Iceberg tables directly. The AWS Glue Data Catalog provides catalog and metadata management for these tables, and Amazon SageMaker builds on the same open Lakehouse foundation, letting analytics and artificial intelligence and machine learning (AI/ML) workloads share a single copy of data across Amazon S3 data lakes (including Amazon S3 Tables), with SageMaker Feature Store supporting an Iceberg offline store for faster training dataset extraction. On the ingestion side, Amazon Data Firehose delivers data into Iceberg tables from sources such as Direct PUT, Amazon Kinesis Data Streams, Amazon Managed Streaming for Apache Kafka (Amazon MSK), Amazon CloudWatch Logs, and AWS IoT, with optional on-the-fly record transformation by using AWS Lambda. No-code Streaming Tables map a single Kinesis Data Streams stream or a single Amazon MSK (Express brokers) Kafka topic to a single Amazon S3 Tables Iceberg table, with the target table auto-created and partitioning configured on the delivery channel.

AWS managed features

AWS provides managed capabilities specifically designed for Iceberg. These managed features reduce operational overhead while maintaining performance and cost efficiency, so teams can focus on deriving insights rather than managing infrastructure.

• AWS Glue Iceberg table optimizer – Automated compaction, snapshot management, and orphan file cleanup without manual intervention on Iceberg tables in General S3 buckets.
• Amazon S3 Tables – Fully managed Iceberg tables with automatic optimization and maintenance.
• Zero-ETL integration – Managed, incremental replication from Amazon DynamoDB and third-party enterprise applications such as Salesforce, SAP, ServiceNow, and Zendesk into Iceberg tables in Amazon S3 (self-managed or Amazon S3 Tables), with no custom pipeline or connector code to maintain. Source changes are applied incrementally, so the target stays in sync without full reloads.
• Iceberg materialized views – SQL-defined views materialized as native Iceberg tables, refreshed on an automatic schedule and incrementally when the source tables allow it, with change detection handled for you. Well suited to consumption-layer tables that feed dashboards, APIs, and downstream pipelines, since the work shifts from query time to the managed refresh.

These benefits make Apache Iceberg a strong choice for modern data lake implementations, offering advantages in performance, manageability, and scalability compared to traditional table formats.

In the context of migrating data to Apache Iceberg, two primary approaches exist: in-place migration and full table migration. We previously covered both in the blog post “Migrate an existing data lake to a transactional data lake using Apache Iceberg” which walks through the methodology for traditional Hive-style tables, meaning partitioned Apache Parquet, Apache ORC, or Apache Avro datasets registered in a Hive-compatible catalog. The same two approaches apply when the starting point is an Apache Hudi data lake, which comes in two flavors, Copy-on-Write (CoW) and Merge-on-Read (MoR).

In-place migration

This approach generates fresh Iceberg metadata on top of the existing data files of the source table, without rewriting the data itself. The Parquet files of a Hive, Hudi, or Delta Lake table stay in place, and a new or existing Iceberg table is pointed at them with its own metadata layer. The main advantage is cost and speed: no data rewrite, so the migration is fast and cheap on both compute and storage. The main limitation is that the Iceberg table inherits the source table’s schema, partitioning, and physical file shape (file sizes, row groups, sort order), and none of these can be changed during the migration itself.

Full table migration

In this approach, the data itself is rewritten into a new Iceberg table, typically using a writeTo() call or a CREATE TABLE AS SELECT (CTAS) statement. It works for any source an engine can read from (Hive, Hudi, Delta Lake, relational databases, and others), and because the data is being rewritten anyway, the migration step is also where you can reshape it: change the schema, apply Iceberg hidden partitioning, resort the data for pruning, and target larger file sizes than the source has. The trade-offs are cost, time, and storage: higher compute spend, longer migration, and the dataset is effectively stored twice (source and target) until the migration completes, and the source is retired. For large historical tables, a single CTAS is usually impractical, so the common pattern is to seed the table with a small CTAS and then run INSERT INTO ... SELECT partition by partition, parallelizing and checkpointing across batches.

In this post we work through both approaches end-to-end. We first create a sample Hudi event table in each flavor, CoW and MoR, then migrate them in place to Iceberg using two techniques, the Spark Iceberg add_files procedure and the Apache XTable sync tool, covering the considerations and prerequisites of each. Finally, we walk through the full table migration path, where the source data is rewritten into a new Iceberg table from scratch.

Solution overview

The following diagram illustrates the architecture that you implement through this post.

Figure 1: Migration architecture showing the in-place paths (add_files and Apache XTable) and the full-table paths (CTAS and INSERT) to Amazon S3 and Amazon S3 Tables, all registered in the AWS Glue Data Catalog and queried with Amazon Athena

The workflow consists of the following steps:

1. An AWS Glue notebook writes the source data as Apache Hudi tables to an Amazon S3 bucket and registers them in the AWS Glue Data Catalog. This creates the existing Hudi data lake that the migration starts from.
2. The first in-place migration approach uses an AWS Glue notebook to run the Apache Iceberg add_files procedure. The procedure creates Iceberg metadata on top of the Parquet data files already in the Hudi table. The new Iceberg metadata points to the same Parquet files, so no data is copied or rewritten. This makes it a fast, low-cost migration path.
3. The second in-place migration approach runs Apache XTable on an Amazon Elastic Compute Cloud (Amazon EC2) instance to convert the Hudi table metadata into Iceberg metadata. XTable reads the Hudi commit timeline and creates matching Iceberg metadata that points to the same Parquet files. Both table formats can then share a single copy of the data.
4. For a full migration, an AWS Glue notebook uses CREATE TABLE AS SELECT (CTAS) or INSERT INTO statements to read the Hudi data and rewrite it as new Iceberg tables. The tables can go in a general purpose S3 bucket or in Amazon S3 Tables. This approach fully separates the new tables from the Hudi source. S3 Tables also handle table maintenance for you, such as compaction and snapshot management.
5. The AWS Glue Data Catalog registers the Iceberg tables from all three approaches. Users can then use Amazon Athena, a serverless query service, to run SQL queries on these tables and validate the migrated data. Athena enables interactive SQL-based exploration of data directly in Amazon S3 without the need for complex infrastructure management.

Prerequisites

Before getting started, make sure you have the following:

• An AWS account with a role that has sufficient access to provision the required resources.
• This walkthrough uses AWS Region us-east-1.
• An AWS Identity and Access Management (IAM) role for your notebook as described in Set up IAM permissions for AWS Glue Studio and Amazon S3 read and write permission.

Launch an AWS Glue Studio notebook for processing

For this post, we use an AWS Glue Studio notebook. Follow the steps in Getting started with notebooks in AWS Glue Studio to set up the notebook environment.

1. Sign in to the AWS Management Console and open the AWS Glue console.
2. In the Region selector (top right), choose your preferred working Region.
3. Choose ETL Jobs in the left pane.
4. Choose Notebook under Author using an interactive code notebook, select Start fresh, choose your IAM role, and choose Create notebook.
5. After a minute, the Jupyter notebook editor appears. Clear all the default cells.

Create Hudi Copy-on-Write (CoW) and Merge-on-Read (MoR) sample event tables

This notebook creates Hudi tables using both Copy-on-Write (CoW) and Merge-on-Read (MoR) table types with sample events. We perform inserts, updates, and deletes, then inspect the underlying file structures to understand the key differences:

• CoW: an update or delete fully rewrites every base Parquet file that contains an impacted row, producing a new file slice in the file group. Reads are cheap, and writes amplify with the fraction of files touched.
• MoR: an update or delete is appended to an Avro log file alongside the base file instead of rewriting the base, and base + logs are merged at read time (or compacted into a new base file later). Cheaper writes. Costlier reads until compaction.

1. Execute the first cell with the following configuration to set up the Spark session with Hudi and Iceberg support:

Code:

{ "glue_version": "5.1", "worker_type": "G.1X", "number_of_workers": 5, "idle_timeout": 30, "--datalake-formats" : "hudi,iceberg", "--conf" : "spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension,org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_iceberg=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_iceberg.warehouse=s3://YOUR_BUCKET/warehouse/ --conf spark.sql.catalog.glue_iceberg.type=glue --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar" }

2. Initialize the session and configuration. Update the bucket_name, db_name and region variables with your configurations. Reusable helper – list_table_files(table_name, label="", verbose=False) renders a table’s S3 layout as a directory tree so you can SEE how CLEAN, COMPACT and add_files change the files underneath. It is reused before/after each of those steps instead of re-coding the listing.

Code:

# ────────────────────────────────────────────── # Source Hudi tables # ────────────────────────────────────────────── hudi_events_cow_table = "events_hudi_cow" # Copy-on-Write source hudi_events_mor_table = "events_hudi_mor" # Merge-on-Read source # ────────────────────────────────────────────── # In-place migration targets (add_files/xtable) # ────────────────────────────────────────────── iceberg_cow_inplace = "events_iceberg_cow_inplace" iceberg_mor_inplace = "events_iceberg_mor_inplace" iceberg_cow_xtable = "events_hudi_cow_iceberg_xtable" iceberg_mor_xtable = "events_hudi_mor_iceberg_xtable" # ────────────────────────────────────────────── # Full table (shadow) migration targets # ────────────────────────────────────────────── iceberg_cow_full_ctas = "events_iceberg_cow_full_ctas" iceberg_cow_full_create_insert = "events_iceberg_cow_full_create_insert" iceberg_mor_full_ctas = "events_iceberg_mor_full_ctas" iceberg_mor_full_create_insert = "events_iceberg_mor_full_create_insert" # ────────────────────────────────────────────── # Infrastructure # ────────────────────────────────────────────── bucket_name = "demoblogiceberg-bdb5001" # S3 bucket for data db_name = "hudi_to_iceberg_blog" # Glue Data Catalog database region = "us-east-1" # Create the Glue Data Catalog database spark.sql(f""" CREATE DATABASE IF NOT EXISTS {db_name} LOCATION 's3://{bucket_name}/{db_name}' """) # Reusable helper —list_table_files()(S3 layout tree) import boto3 s3 = boto3.client('s3', region_name=region) # File/folder annotations keyed by classifier result _NOTES = { "hoodie_dir": "<-- Hudi metadata (timeline, schema, indexes)", "hoodie_mdt": "<-- Hudi metadata table (internal index & stats, MoR)", "hoodie_files": "<-- file listing index", "hoodie_cstats":"<-- column-level min/max/null stats", "hoodie_ridx": "<-- record-key -> file-group mapping", "ice_meta_dir": "<-- Iceberg metadata layer (metadata.json, manifest lists, manifests)", "ice_data_dir": "<-- Iceberg data layer (base Parquet files)", "base": "<-- base Parquet file", "log": "<-- Hudi delta log file (MoR update, uncompacted)", "part_meta": "<-- Hudi partition marker", "metadata_json":"<-- Iceberg table metadata (schema, snapshots)", "manifest_list":"<-- Iceberg manifest list (snapshot -> manifests)", "manifest": "<-- Iceberg manifest (lists data files)", "props": "<-- table configuration & schema", } # Fixed-path annotations for metadata directories _DIR_KIND = { ".hoodie": "hoodie_dir", ".hoodie/metadata": "hoodie_mdt", ".hoodie/metadata/files": "hoodie_files", ".hoodie/metadata/column_stats": "hoodie_cstats", ".hoodie/metadata/record_index": "hoodie_ridx", "metadata": "ice_meta_dir", "data": "ice_data_dir", } def _fmt_size(n): """Human-readable byte size.""" for u in ("B", "KB", "MB", "GB", "TB"): if n < 1024 or u == "TB": return f"{n:,.0f} {u}" if u == "B" else f"{n:,.1f} {u}" n /= 1024.0 def _leaf_kind(name): """Classify a leaf file by name -> annotation key / count bucket.""" if ".log." in name: return "log" if name.startswith(".hoodie_partition_metadata"): return "part_meta" if name == "hoodie.properties": return "props" if name.endswith(".metadata.json"): return "metadata_json" if name.startswith("snap-") and name.endswith(".avro"): return "manifest_list" if name.endswith(".avro"): return "manifest" if name.endswith(".parquet"): return "base" return "" def list_table_files(table_name, label="", verbose=False, max_depth=2, max_children=2): """Render a Hudi/Iceberg table's S3 layout as an annotated directory tree. Teaching view for CLEAN/COMPACT/add_files: multiple base files in a partition = stale file slices; .log.* = uncompacted MoR updates. The large Hudi .hoodie/ is summarized (depth/width capped); Iceberg metadata/ and data partitions show in full. verbose=True expands everything. """ prefix = f"{db_name}/{table_name}/" header = f"s3://{bucket_name}/{prefix}" if label: header += f" [{label}]" print(header) in_meta = lambda p: (p.split("/", 1)[0] if p else "") in (".hoodie", "metadata") capped_meta = lambda p: (p.split("/", 1)[0] if p else "") == ".hoodie" # ---- gather objects into a nested tree + tally honest counts ----------- root = {"d": {}, "f": []} n_base = n_log = n_meta = n_other = 0 s_base = s_log = s_meta = 0 has_hudi = has_ice = False for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucket_name, Prefix=prefix): for obj in page.get("Contents", []): key = obj["Key"].replace(prefix, "") size = obj["Size"] if key == "" or key.endswith("/"): continue # S3 directory marker parts = key.split("/") fname, top = parts[-1], parts[0] if top == ".hoodie": has_hudi = True if top == "metadata": has_ice = True # files inside a metadata folder count ONLY as metadata if top in (".hoodie", "metadata"): n_meta += 1; s_meta += size elif ".log." in fname: n_log += 1; s_log += size elif fname.endswith(".parquet"): n_base += 1; s_base += size elif fname.startswith(".hoodie"): n_meta += 1; s_meta += size; has_hudi = True else: n_other += 1 node = root for part in parts[:-1]: node = node["d"].setdefault(part, {"d": {}, "f": []}) node["f"].append((fname, size)) # ---- helpers ----------------------------------------------------------- def count_all(node): return len(node["f"]) + sum(count_all(c) for c in node["d"].values()) def collapse(node, name, rel): """Merge single-child, file-less dir chains (year=/month=/day=).""" while (len(node["d"]) == 1 and not node["f"] and not in_meta(rel)): cname, child = next(iter(node["d"].items())) crel = f"{rel}/{cname}" if in_meta(crel): break name, rel, node = f"{name}/{cname}", crel, child return name, rel, node def line(indent, branch, text, note, width): s = f"{indent}{branch}{text}" return f"{s:<{width}} {note}" if note else s def render(node, rel, indent, depth): entries = ([("d", n, c) for n, c in sorted(node["d"].items())] + [("f", n, sz) for n, sz in sorted(node["f"], key=lambda f: (".log." in f[0], f[0]))]) cap = (not verbose) and capped_meta(rel) and len(entries) > max_children shown = entries[:max_children] if cap else entries hidden = len(entries) - len(shown) for i, item in enumerate(shown): last = (i == len(shown) - 1) and hidden == 0 branch = "└── " if last else "├── " sub = indent + (" " if last else "│ ") if item[0] == "d": _, name, child = item disp, erel, enode = collapse(child, name, f"{rel}/{name}" if rel else name) note = _NOTES.get(_DIR_KIND.get(erel, ""), "") cnt = f" ({count_all(enode)} obj)" if (in_meta(erel) or erel in _DIR_KIND) else "" print(line(indent, branch, f"{disp}/{cnt}", note, 60)) if (not verbose) and capped_meta(erel) and depth + 1 > max_depth: inner = count_all(enode) if inner: print(f"{sub}└── … ({inner} objects, use verbose=True)") else: render(enode, erel, sub, depth + 1) else: _, name, size = item note = _NOTES.get(_leaf_kind(name), "") print(line(indent, branch, f"{name} ({_fmt_size(size)})", note, 72)) if hidden: print(f"{indent}└── … (+{hidden} more)") render(root, "", "", 0) # ---- per-partition slice note (Hudi concept; skip metadata & Iceberg data/) ---- def slices(node, rel): top = rel.split("/", 1)[0] if rel else "" if top in (".hoodie", "metadata", "data"): return nb = sum(1 for f, _ in node["f"] if f.endswith(".parquet")) nl = sum(1 for f, _ in node["f"] if ".log." in f) if nb > 1: print(f" note: {rel}/ has {nb} base files -> multiple file slices (stale versions)") elif nb == 1 and nl: print(f" note: {rel}/ has 1 base + {nl} log file(s) -> uncompacted MoR updates") for n, c in node["d"].items(): slices(c, f"{rel}/{n}" if rel else n) slices(root, "") # ---- in-place migration note ------------------------------------------- if has_ice and "data" not in root["d"] and n_base == 0: print(" note: Iceberg metadata/ present but no local data/ folder") print(" -> in-place migration: data files remain registered in place at the source (Hudi) location; no data copied") # ---- summary ----------------------------------------------------------- total = n_base + n_log + n_meta + n_other print( f"SUMMARY: {n_base} base · {n_log} log · {n_meta} meta" + (f" · {n_other} other" if n_other else "") + f" · {total} files · {_fmt_size(s_base + s_log + s_meta)}" ) print(f" .hoodie/: {'present' if has_hudi else 'none'} metadata/ (Iceberg): {'present' if has_ice else 'none'}") print()

Sample event data

We use a website clickstream events dataset, partitioned by year/month/day – a realistic pattern for event-driven architectures.

Note: In this example, we create the table with a preCombineField, which flips the default write operation of INSERT INTO from insert to upsert. You can override it explicitly with hoodie.spark.sql.insert.into.operation when you need a different behavior.

CoW table: create, upsert, and delete

We create a Hudi CoW table, insert records, apply an upsert, and delete a record, then inspect how the files change.

Create the Hudi CoW events table

Code:

spark.sql(f""" CREATE TABLE {db_name}.{hudi_events_cow_table} ( event_id STRING, user_id STRING, event_type STRING, event_timestamp TIMESTAMP, page_url STRING, year INT, month INT, day INT ) USING HUDI OPTIONS ( type = 'cow', primaryKey = 'event_id', preCombineField = 'event_timestamp' ) PARTITIONED BY (year, month, day) LOCATION 's3://{bucket_name}/{db_name}/{hudi_events_cow_table}/' """)

Insert event records

Code:

spark.sql(f""" INSERT INTO {db_name}.{hudi_events_cow_table} VALUES ('EVT_001', 'USER_101', 'page_view', timestamp('2024-04-01 08:15:00'), '/home', 2024, 4, 1), ('EVT_002', 'USER_102', 'click', timestamp('2024-04-01 09:30:00'), '/products', 2024, 4, 1), ('EVT_003', 'USER_103', 'purchase', timestamp('2024-04-02 10:45:00'), '/checkout', 2024, 4, 2), ('EVT_004', 'USER_101', 'page_view', timestamp('2024-04-02 14:20:00'), '/blog', 2024, 4, 2), ('EVT_005', 'USER_104', 'signup', timestamp('2024-04-03 11:00:00'), '/register', 2024, 4, 3), ('EVT_006', 'USER_102', 'click', timestamp('2024-04-03 16:45:00'), '/deals', 2024, 4, 3) """)

Upsert: simulate late-arriving corrections

Code:

spark.sql(f""" INSERT INTO {db_name}.{hudi_events_cow_table} VALUES ('EVT_002', 'USER_102', 'add_to_cart', timestamp('2024-04-01 09:31:00'), '/products/item-42', 2024, 4, 1), ('EVT_005', 'USER_104', 'signup', timestamp('2024-04-03 11:01:00'), '/welcome', 2024, 4, 3) """)

Delete: remove a record

Code:

spark.sql(f""" DELETE FROM {db_name}.{hudi_events_cow_table} WHERE event_id = 'EVT_006' """)

Inspect the CoW file structure

CoW tables rewrite the entire Parquet file on every upsert/delete. You will see one base Parquet file per file group (no log files).

Code:

list_table_files(hudi_events_cow_table, "COW --- initial file structure")

Output:

s3://demoblogiceberg-bdb5001/hudi_to_iceberg_blog/events_hudi_cow/ [COW — initial file structure] ├── .hoodie/ (50 obj) <-- Hudi metadata (timeline, schema, indexes) │ ├── .......trimmed └── year=2024/month=4/ ├── day=1/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ ├── 7d187e9a.....20261011152509633.parquet (426.3 KB) <-- base Parquet file │ └── 7d187e9a.....20261011152403779.parquet (426.4 KB) <-- base Parquet file ├── day=2/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ └── e583fe17.....20261011152403779.parquet (426.4 KB) <-- base Parquet file └── day=3/ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker ├── eedc1b38.....20261011152509633.parquet (426.3 KB) <-- base Parquet file ├── eedc1b38.....20261011152539102.parquet (426.2 KB) <-- base Parquet file └── eedc1b38.....20261011152403779.parquet (426.4 KB) <-- base Parquet file note: year=2024/month=4/day=1/ has 2 base files -> multiple file slices (stale versions) note: year=2024/month=4/day=3/ has 3 base files -> multiple file slices (stale versions) SUMMARY: 6 base · 0 log · 53 meta · 59 files · 2.8 MB .hoodie/: present metadata/ (Iceberg): none

MoR table: create, upsert, and delete

We run the same create, upsert, and delete sequence on a Hudi MoR table.

Create the Hudi MoR events table

Code:

spark.sql(f""" CREATE TABLE {db_name}.{hudi_events_mor_table} ( event_id STRING, user_id STRING, event_type STRING, event_timestamp TIMESTAMP, page_url STRING, year INT, month INT, day INT ) USING HUDI OPTIONS ( type = 'mor', primaryKey = 'event_id', preCombineField = 'event_timestamp' ) PARTITIONED BY (year, month, day) LOCATION 's3://{bucket_name}/{db_name}/{hudi_events_mor_table}/' """)

Insert initial event records

Code:

spark.sql(f""" INSERT INTO {db_name}.{hudi_events_mor_table} VALUES ('EVT_001', 'USER_201', 'page_view', timestamp('2024-05-10 07:00:00'), '/home', 2024, 5, 10), ('EVT_002', 'USER_202', 'click', timestamp('2024-05-10 08:15:00'), '/products', 2024, 5, 10), ('EVT_003', 'USER_203', 'purchase', timestamp('2024-05-11 09:30:00'), '/checkout', 2024, 5, 11), ('EVT_004', 'USER_201', 'page_view', timestamp('2024-05-11 12:00:00'), '/account', 2024, 5, 11), ('EVT_005', 'USER_204', 'signup', timestamp('2024-05-12 10:30:00'), '/register', 2024, 5, 12), ('EVT_006', 'USER_202', 'click', timestamp('2024-05-12 15:45:00'), '/deals', 2024, 5, 12) """)

Upsert: simulate late-arriving corrections (creates delta log files)

Code:

spark.sql(f""" INSERT INTO {db_name}.{hudi_events_mor_table} VALUES ('EVT_002', 'USER_202', 'add_to_cart', timestamp('2024-05-10 08:16:00'), '/products/item-99', 2024, 5, 10), ('EVT_005', 'USER_204', 'signup', timestamp('2024-05-12 10:31:00'), '/welcome', 2024, 5, 12) """)

Delete: remove a record

Code:

spark.sql(f""" DELETE FROM {db_name}.{hudi_events_mor_table} WHERE event_id = 'EVT_006' """)

Inspect MoR: base Parquet and log files

MoR tables write updates as delta log files (.log.*) alongside the base Parquet files. This avoids full file rewrites for faster writes.

Code:

list_table_files(hudi_events_mor_table, "MOR --- base parquet + log files")

Output:

s3://demoblogiceberg-bdb5001/hudi_to_iceberg_blog/events_hudi_mor/ [MOR — base parquet + log files] ├── .hoodie/ (50 obj) <-- Hudi metadata (timeline, schema, indexes) │ ├── .......trimmed └── year=2024/month=5/ ├── day=10/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ ├── 7d5ce247.....20261011153110535.parquet (426.4 KB) <-- base Parquet file │ └── .7d5ce247.....20261011153145862.log.1_1-164-788 (1.3 KB) <-- Hudi delta log file (MoR update, uncompacted) ├── day=11/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ └── f9b8147e.....20261011153110535.parquet (426.4 KB) <-- base Parquet file └── day=12/ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker ├── 3f0b5ad4.....20261011153110535.parquet (426.4 KB) <-- base Parquet file ├── .3f0b5ad4.....20261011153145862.log.1_0-164-787 (1.3 KB) <-- Hudi delta log file (MoR update, uncompacted) └── .3f0b5ad4.....20261011153215018.log.1_0-192-901 (1.1 KB) <-- Hudi delta log file (MoR update, uncompacted) note: year=2024/month=5/day=10/ has 1 base + 1 log file(s) -> uncompacted MoR updates note: year=2024/month=5/day=12/ has 1 base + 2 log file(s) -> uncompacted MoR updates SUMMARY: 3 base · 3 log · 53 meta · 59 files · 1.5 MB .hoodie/: present metadata/ (Iceberg): none

Query both Hudi tables

Code:

print("=== COW Table (events_hudi_cow) ===") spark.sql(f"SELECT event_id, user_id, event_type, event_timestamp, page_url, year, month, day FROM {db_name}.{hudi_events_cow_table} ORDER BY event_id").show(truncate=False) print("\n=== MOR Table (events_hudi_mor) ===") spark.sql(f"SELECT event_id, user_id, event_type, event_timestamp, page_url, year, month, day FROM {db_name}.{hudi_events_mor_table} ORDER BY event_id").show(truncate=False)

Output:

=== COW Table (events_hudi_cow) === +--------+--------+-----------+-------------------+-----------------+----+-----+---+ |event_id|user_id |event_type |event_timestamp |page_url |year|month|day| +--------+--------+-----------+-------------------+-----------------+----+-----+---+ |EVT_001 |USER_101|page_view |2024-04-01 08:15:00|/home |2024|4 |1 | |EVT_002 |USER_102|add_to_cart|2024-04-01 09:31:00|/products/item-42|2024|4 |1 | |EVT_003 |USER_103|purchase |2024-04-02 10:45:00|/checkout |2024|4 |2 | |EVT_004 |USER_101|page_view |2024-04-02 14:20:00|/blog |2024|4 |2 | |EVT_005 |USER_104|signup |2024-04-03 11:01:00|/welcome |2024|4 |3 | +--------+--------+-----------+-------------------+-----------------+----+-----+---+ === MOR Table (events_hudi_mor) === +--------+--------+-----------+-------------------+-----------------+----+-----+---+ |event_id|user_id |event_type |event_timestamp |page_url |year|month|day| +--------+--------+-----------+-------------------+-----------------+----+-----+---+ |EVT_001 |USER_201|page_view |2024-05-10 07:00:00|/home |2024|5 |10 | |EVT_002 |USER_202|add_to_cart|2024-05-10 08:16:00|/products/item-99|2024|5 |10 | |EVT_003 |USER_203|purchase |2024-05-11 09:30:00|/checkout |2024|5 |11 | |EVT_004 |USER_201|page_view |2024-05-11 12:00:00|/account |2024|5 |11 | |EVT_005 |USER_204|signup |2024-05-12 10:31:00|/welcome |2024|5 |12 | +--------+--------+-----------+-------------------+-----------------+----+-----+---+

Migrate Hudi tables to Iceberg with in-place migration

In-place migration keeps the original Parquet data files where they are and only builds Iceberg metadata on top of them. There is no rewrite of the underlying data, which makes the migration fast and inexpensive even on large tables. Two tools cover the common cases on AWS:

• add_files (Iceberg built-in procedure). A Spark procedure shipped with Apache Iceberg that registers existing Parquet files into an Iceberg table by writing the corresponding manifests, without rewriting the data. It runs as a Spark job in which the source table’s partitions are distributed across Spark tasks, so file listing and column-statistics gathering happen in parallel on the cluster.
• Apache XTable (incubating). An open-source cross-format translation tool that can read the metadata of one lakehouse format (Hudi, Delta, or Iceberg) and write out the equivalent metadata for another, pointing at the same underlying Parquet files. XTable can run once to produce the initial Iceberg metadata, or run on a schedule to keep the Iceberg view continuously in sync with ongoing writes to the Hudi table.

Choosing between add_files and Apache XTable for in-place migration

When planning an in-place migration, it is important to understand the strengths of each approach:

• add_files is best suited for a one-time migration. It registers existing Hudi Parquet files into an Iceberg table in a single operation. After the migration is complete, the source Hudi table and target Iceberg table are independent, and any subsequent changes to the Hudi table are not reflected in Iceberg.
• Apache XTable supports both an initial one-time conversion and ongoing incremental synchronization. This makes it ideal for transition periods where you need to keep both the source Hudi tables and the target Iceberg tables in sync while downstream consumers gradually migrate to Iceberg. After all workloads have been cut over, you can stop the sync process and decommission the Hudi tables.

Important notes

add_files/XTable register existing Parquet files as-is, so they rely on the column names baked into each file. Added and dropped columns are safe, but if the source table was renamed any column during its lifecycle, files written before the rename still hold the old name and reads against the new name return NULL for them. For tables with a history of column renames, prefer a full migration instead.

Iceberg version and column statistics (in-place add_files / snapshot): In Iceberg 1.9.0 through 1.10.0, the add_files and snapshot procedures don’t generate column stats in the Iceberg metadata. The cause is a NameMapping loss in ParquetUtil.footerMetrics. Without those column-level lower/upper bounds and null counts, the migrated Iceberg table loses most of its file-skipping ability at query planning time, which translates into more Parquet files being inspected, higher scan volume, and slower queries, especially for selective filters on non-partition columns. The issue is fixed in the Iceberg 1.10.1 release (#14617). Affected runtimes are AWS Glue 5.1 and Amazon EMR 7.11–7.13. We recommend AWS Glue 6.0+ or Amazon EMR 7.14+ for production migrations.

Shared data files during in-place migration: With both approaches, the in-place migration doesn’t copy data. The resulting Iceberg table points to the same underlying Parquet files as the source Hudi table. This means that until you perform a full cutover to Iceberg, you must avoid destructive operations on the Iceberg table that could delete or modify the shared data files. Specifically:

• Don’t drop the Iceberg table with the PURGE option.
• Don’t run Iceberg maintenance tasks such as expire_snapshots or remove_orphan_files.

Running any of these before cutover can remove files that the source Hudi table still depends on, leading to data loss. Iceberg exposes a built-in guardrail against this: the gc.enabled table property. When set to false, Iceberg blocks the preceding destructive operations. How you apply this depends on which migration path you choose.

With add_files. Disable garbage collection immediately after the procedure finishes and keep it disabled until you have fully cut over to Iceberg.

Right after add_files:

spark.sql(f""" ALTER TABLE {db_name}.{table_name} SET TBLPROPERTIES ('gc.enabled' = 'false'); """)

After Hudi source is retired and the Iceberg table owns its own data, re-enable GC so maintenance can run normally:

spark.sql(f""" ALTER TABLE {db_name}.{table_name} SET TBLPROPERTIES ('gc.enabled' = 'true'); """)

With Apache XTable (incremental sync). The gc.enabled = false guardrail is only partially usable here, because XTable’s incremental sync needs GC to be enabled on the Iceberg table to operate normally. A practical pattern is:

1. Run the initial full XTable sync.
2. Temporarily set gc.enabled = false to run post-migration validation and consistency checks on the Iceberg table in a safe state.
3. Set gc.enabled = true again before starting incremental sync and leave it on for the duration of the sync period.

Because the automatic guardrail is off during incremental sync, the burden shifts to operational discipline: as long as XTable is syncing and data files are shared with the Hudi table, don’t run any Iceberg operation that can delete source data files (no DROP TABLE ... PURGE, no expire_snapshots, no remove_orphan_files, no manual object deletion against the shared S3 prefix). Treat the Iceberg table as read-only in terms of file lifecycle until cutover is complete, and Hudi is decommissioned, at which point normal Iceberg maintenance can resume.

Don’t enable the AWS Glue Data Catalog table optimizer before cutover.

Glue-managed Iceberg maintenance (compaction, snapshot retention, orphan file cleanup) will rewrite or delete the Parquet files that the source Hudi table still depends on. Verify that no catalog-level optimizer applies to the migrated table and leave table-level optimization off until cutover is complete.

Prerequisites for in-place migration with add_files. The add_files procedure registers every Parquet file it finds under the source location into the target Iceberg table, with no awareness of Hudi’s file-group versioning. If more than one file version of the same record is present at migration time, each version is registered as a separate row, and the Iceberg table will contain duplicates. The source Hudi table must therefore expose exactly one file version per record before add_files runs:

• CoW tables: Run CLEAN to remove old file versions, keeping only the latest Parquet files.
• MoR tables: Run COMPACTION (merge delta logs into base Parquet) + CLEAN before calling add_files, since add_files can only register base Parquet files and cannot process Hudi delta log files.

Option 1: Migrate Hudi tables to Iceberg with the add_files Spark procedure

This option uses the Iceberg add_files procedure to register the existing Hudi Parquet files into new Iceberg tables. Run it once per table, after preparing the Hudi table so that only the current file versions remain.

Migrate CoW to Iceberg

We create the target Iceberg table, prepare the Hudi source so that only the current file versions remain, register those files, and validate the result.

Create the target Iceberg table with the same schema

Code:

spark.sql(f""" CREATE TABLE glue_iceberg.{db_name}.{iceberg_cow_inplace} ( event_id STRING, user_id STRING, event_type STRING, event_timestamp TIMESTAMP, page_url STRING, year INT, month INT, day INT ) USING iceberg PARTITIONED BY (year, month, day) LOCATION 's3://{bucket_name}/{db_name}/{iceberg_cow_inplace}/' """)

Run Hudi CLEAN to keep only the latest file versions

Check files listing for the Hudi table before the CLEAN operation:

Code:

list_table_files(hudi_events_cow_table, "COW --- BEFORE CLEAN")

Output:

s3://demoblogiceberg-bdb5001/hudi_to_iceberg_blog/events_hudi_cow/ [COW — BEFORE CLEAN] ├── .hoodie/ (50 obj) <-- Hudi metadata (timeline, schema, indexes) │ ├── .......trimmed └── year=2024/month=4/ ├── day=1/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ ├── 7d187e9a.....20261011152509633.parquet (426.3 KB) <-- base Parquet file │ └── 7d187e9a.....20261011152403779.parquet (426.4 KB) <-- base Parquet file ├── day=2/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ └── e583fe17.....20261011152403779.parquet (426.4 KB) <-- base Parquet file └── day=3/ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker ├── eedc1b38.....20261011152509633.parquet (426.3 KB) <-- base Parquet file ├── eedc1b38.....20261011152539102.parquet (426.2 KB) <-- base Parquet file └── eedc1b38.....20261011152403779.parquet (426.4 KB) <-- base Parquet file note: year=2024/month=4/day=1/ has 2 base files -> multiple file slices (stale versions) note: year=2024/month=4/day=3/ has 3 base files -> multiple file slices (stale versions) SUMMARY: 6 base · 0 log · 53 meta · 59 files · 2.8 MB .hoodie/: present metadata/ (Iceberg): none

Run the CLEAN operation specifying to keep only the latest file slices:

Code:

spark.sql(f""" CALL spark_catalog.system.run_clean( table => '{db_name}.{hudi_events_cow_table}', clean_policy => 'KEEP_LATEST_FILE_VERSIONS', file_versions_retained => 1 ) """).show(truncate=False)

Output:

+-----------------+--------------------+-------------------+-------------------------+-----------------------+-------+ |start_clean_time |time_taken_in_millis|total_files_deleted|earliest_commit_to_retain|bootstrap_part_metadata|version| +-----------------+--------------------+-------------------+-------------------------+-----------------------+-------+ |20261011154407413|382 |3 | |{} |2 | +-----------------+--------------------+-------------------+-------------------------+-----------------------+-------+

Check files listing for the Hudi table after the CLEAN operation:

Code:

list_table_files(hudi_events_cow_table, "COW --- AFTER CLEAN")

Output:

s3://demoblogiceberg-bdb5001/hudi_to_iceberg_blog/events_hudi_cow/ [COW — AFTER CLEAN] ├── .hoodie/ (59 obj) <-- Hudi metadata (timeline, schema, indexes) │ ├── .......trimmed └── year=2024/month=4/ ├── day=1/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ └── 7d187e9a.....20261011152509633.parquet (426.3 KB) <-- base Parquet file ├── day=2/ │ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker │ └── e583fe17.....20261011152403779.parquet (426.4 KB) <-- base Parquet file └── day=3/ ├── .hoodie_partition_metadata (96 B) <-- Hudi partition marker └── eedc1b38.....20261011152539102.parquet (426.2 KB) <-- base Parquet file SUMMARY: 3 base · 0 log · 62 meta · 65 files · 1.6 MB .hoodie/: present metadata/ (Iceberg): none

Register Hudi Parquet files into Iceberg with add_files

Code:

spark.sql(f""" CALL glu...