Approach (step-by-step)1. Scope & baseline: record source parquet layout (partition columns, file paths, file counts, per-file row counts, min/max/null stats from footers).2. Create Iceberg table with identical schema and partition spec. Ingest data (bulk load or rewrite) without altering partition keys.3. Validate at three levels: file-level, partition-level, and logical table-level (rows & column checksums). Also validate Iceberg metadata (manifests, snapshots, partition metrics).Verification queries & commandsA. Source (Parquet) inventory (Spark)scala
// list files, sizes, row counts via Spark
spark.read.format("parquet").load("s3://bucket/source/path")
.selectExpr("input_file_name() as file", "size", "count(*) over () as total_rows")
.groupBy("file").count().show(false)
sql
-- Source: read raw parquet
SELECT COUNT(*) AS src_rows FROM parquet.`s3://bucket/source/path`;
-- Target: Iceberg table
SELECT COUNT(*) AS tgt_rows FROM iceberg_db.table;
sql
-- Source
SELECT dt, COUNT(*) AS rows, COUNT(DISTINCT(input_file_name())) AS files
FROM parquet.`s3://bucket/source/path`
GROUP BY dt;
-- Target
SELECT dt, SUM(record_count) AS rows, COUNT(*) AS manifests_or_files
FROM iceberg_db.table.snapshots -- or use table.files() API
GROUP BY dt;
sql
-- example using Spark SQL
SELECT md5(concat_ws('|', col1, col2, col3)) AS row_hash
FROM parquet.`s3://bucket/source/path`
-- aggregate
SELECT count(*) AS cnt, md5(concat_ws(',', collect_list(row_hash))) AS partition_hash
FROM (...)
GROUP BY dt;
sql
SELECT dt, COUNT(*) AS cnt,
md5(concat_ws(',', collect_list(md5(concat_ws('|', col1,col2,col3))))) AS partition_hash
FROM iceberg_db.table
GROUP BY dt;
sql
-- Source min/max/nulls
SELECT dt,
MIN(colA) AS minA, MAX(colA) AS maxA,
SUM(CASE WHEN colA IS NULL THEN 1 ELSE 0 END) AS nullsA
FROM parquet.`...` GROUP BY dt;
-- Target same query against iceberg table
python
from iceberg.api import Table
t = catalog.load_table("db.table")
for mf in t.currentSnapshot().allManifests():
for entry in mf:
print(entry.file_path, entry.file_size, entry.record_count)
