Loading data into Iceberg: appends, overwrites and row-level SQL

# 1. append: add data, touch nothing existing
df.writeTo("ice.blog_ingestion.events").append()

# 2. dynamic partition overwrite: replace only the partitions present in df
df.writeTo("ice.blog_ingestion.events").overwritePartitions()

# 3. filtered overwrite: replace exactly what the filter matches
df.writeTo("ice.blog_ingestion.events").overwrite(F.col("event_ts") >= F.lit("2026-01-07").cast("timestamp"))

# 4. row-level SQL
spark.sql("UPDATE ice.blog_ingestion.events SET region = 'r9' WHERE id < 100")
spark.sql("DELETE FROM ice.blog_ingestion.events WHERE region = 'test'")
spark.sql("MERGE INTO ice.blog_ingestion.events t USING changes s ON t.id = s.id ...")

w = Window.partitionBy("id").orderBy(F.col("log_seq").desc())
latest = changes.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")

(df.repartition(24, F.to_date("event_ts"))   # the days(event_ts) partition value
   .sortWithinPartitions("region")
   .writeTo("ice.blog_ingestion.events").append())

(daily_df.writeTo("ice.blog_ingestion.events").overwritePartitions())

if daily_df.head(1):
    daily_df.writeTo("ice.blog_ingestion.events").overwritePartitions()
else:
    raise ValueError("refusing to run a load that produced no rows")

(daily_df.writeTo("ice.blog_ingestion.events")
         .overwrite(F.to_date("event_ts") == F.lit("2026-01-06").cast("date")))

"""Append, dynamic and filtered overwrite, and row-level SQL on an Iceberg events table.

Run inside the datalake-playground stack:
  docker cp ingestion_modes.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/ingestion_modes.py'
"""
from datetime import datetime, timedelta

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder.appName("iceberg-ingestion-modes")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
T = "ice.blog_ingestion.events"
REGIONS = ["us-east", "us-west", "eu-west", "ap-south"]


def events(day, n=1000, id_start=0, region=None):
    """n events on one day, spread over the day, amounts deterministic."""
    base = datetime(2026, 1, day)
    rows = [(id_start + i,
             base + timedelta(seconds=i * 86_400 // n),
             region or REGIONS[i % 4],
             round(10 + (i % 97) * 1.5, 2))
            for i in range(n)]
    return (spark.createDataFrame(rows, "id BIGINT, event_ts TIMESTAMP, region STRING, amount DOUBLE")
                 .withColumn("amount_usd", F.col("amount").cast("decimal(10,2)")).drop("amount"))


def last_snapshot(table=T):
    r = sql(f"""SELECT snapshot_id, operation, summary['added-records'] AS added,
                       summary['deleted-records'] AS deleted, summary['added-delete-files'] AS del_files,
                       summary['total-records'] AS total
                FROM {table}.snapshots ORDER BY committed_at DESC LIMIT 1""").collect()[0]
    return r


def show(label, table=T):
    r = last_snapshot(table)
    rows = sql(f"SELECT count(*) FROM {table}").collect()[0][0]
    print(f"{label:<44} op={r.operation:<9} added={r.added} deleted={r.deleted} "
          f"delete_files={r.del_files} rows={rows}")


def snapshot_count(table=T):
    return sql(f"SELECT count(*) FROM {table}.snapshots").collect()[0][0]


def per_day(table=T):
    return {str(r.d): r.n for r in sql(
        f"SELECT to_date(event_ts) AS d, count(*) AS n FROM {table} GROUP BY 1 ORDER BY 1").collect()}


sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_ingestion")
sql(f"DROP TABLE IF EXISTS {T} PURGE")
sql(f"""CREATE TABLE {T} (id BIGINT, event_ts TIMESTAMP, region STRING, amount_usd DECIMAL(10,2))
        USING iceberg PARTITIONED BY (days(event_ts))""")

print("== 1. the four ways in ==")
days = events(5, id_start=0)
for d in range(6, 10):
    days = days.unionByName(events(d, id_start=(d - 5) * 1000))
days.writeTo(T).append()
show("append() of 5 days")

events(6, n=800, id_start=90_000).writeTo(T).overwritePartitions()
show("overwritePartitions() with day 6 only")
print("   rows per day:", per_day())

(events(7, n=600, id_start=91_000).writeTo(T)
    .overwrite(F.col("event_ts") >= F.lit("2026-01-07").cast("timestamp")))
show("overwrite(event_ts >= 2026-01-07) with day 7")
print("   rows per day:", per_day())

print("\n== 2. the empty DataFrame ==")
empty = events(8, n=1).limit(0)
before = snapshot_count()
empty.writeTo(T).overwritePartitions()
print("overwritePartitions() of an empty DataFrame: snapshots", before, "->", snapshot_count(),
      "| rows", sql(f"SELECT count(*) FROM {T}").collect()[0][0])
empty.writeTo(T).overwrite(F.to_date("event_ts") == F.lit("2026-01-06").cast("date"))
show("overwrite(day 6) with an empty DataFrame")
print("   rows per day:", per_day())

print("\n== 3. row-level SQL, copy-on-write (the default) ==")
events(8, id_start=8000).unionByName(events(9, id_start=9000)).writeTo(T).append()
sql(f"UPDATE {T} SET region = 'r9' WHERE id < 100")
show("UPDATE ... WHERE id < 100")
sql(f"DELETE FROM {T} WHERE region = 'us-west' AND to_date(event_ts) = DATE'2026-01-08'")
show("DELETE rows inside day 8")
sql(f"DELETE FROM {T} WHERE event_ts >= TIMESTAMP'2026-01-09 00:00:00'")
show("DELETE all of day 9 (whole files)")

print("\n== 4. row-level SQL, merge-on-read ==")
sql(f"""ALTER TABLE {T} SET TBLPROPERTIES ('write.delete.mode'='merge-on-read',
        'write.update.mode'='merge-on-read', 'write.merge.mode'='merge-on-read')""")
sql(f"DELETE FROM {T} WHERE region = 'eu-west' AND to_date(event_ts) = DATE'2026-01-08'")
show("DELETE rows inside day 8, merge-on-read")
sql(f"UPDATE {T} SET region = 'r8' WHERE region = 'ap-south' AND to_date(event_ts) = DATE'2026-01-08'")
show("UPDATE rows inside day 8, merge-on-read")

print("\n== 5. format v1 ==")
V1 = "ice.blog_ingestion.events_v1"
sql(f"DROP TABLE IF EXISTS {V1} PURGE")
sql(f"""CREATE TABLE {V1} (id BIGINT, event_ts TIMESTAMP, region STRING, amount_usd DECIMAL(10,2))
        USING iceberg PARTITIONED BY (days(event_ts)) TBLPROPERTIES ('format-version'='1')""")
events(5).writeTo(V1).append()
sql(f"UPDATE {V1} SET region = 'r9' WHERE id < 100")
show("v1 UPDATE", V1)
sql(f"DELETE FROM {V1} WHERE region = 'us-west'")
show("v1 DELETE", V1)

print("\n== 6. what to check after a load ==")
sql(f"""SELECT operation, summary['added-data-files'] AS files,
               summary['added-records'] AS records, summary['deleted-records'] AS deleted
        FROM {T}.snapshots ORDER BY committed_at""").show(truncate=False)
spark.stop()

"""MERGE INTO, duplicate keys, output file counts, writing by path, add_files and CTAS.

Run inside the datalake-playground stack:
  docker cp merge_and_files.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/merge_and_files.py'
"""
from datetime import datetime, timedelta

from pyspark.sql import SparkSession, Window, functions as F

spark = (
    SparkSession.builder.appName("iceberg-merge-and-files")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
NS = "ice.blog_ingestion"
SRC = "s3a://warehouse/blog_ingestion_src"


def err(e):
    j = getattr(e, "java_exception", None)
    if j is None:
        msg = str(e).splitlines()[0]
        return msg if len(msg) <= 110 else msg[:107] + "..."
    while j.getCause() is not None:             # a failed task wraps the real error
        j = j.getCause()
    msg = j.getClass().getSimpleName() + ": " + j.getMessage().splitlines()[0]
    return msg if len(msg) <= 110 else msg[:107] + "..."


sql(f"CREATE NAMESPACE IF NOT EXISTS {NS}")

print("== 1. MERGE INTO applying a changelog ==")
TGT = f"{NS}.tgt"
sql(f"DROP TABLE IF EXISTS {TGT} PURGE")
sql(f"CREATE TABLE {TGT} (id BIGINT, v STRING, op STRING) USING iceberg")
spark.range(0, 1000).selectExpr("id", "concat('v', id) AS v", "'init' AS op").writeTo(TGT).append()
# 1,000 changes: ids 500..1499. Of the 500 that match, 200 are deletes and 300 updates.
changes = spark.range(500, 1500).selectExpr(
    "id", "concat('new', id) AS v",
    "CASE WHEN id < 700 THEN 'D' WHEN id < 1000 THEN 'U' ELSE 'I' END AS chg")
changes.createOrReplaceTempView("changes")
sql(f"""MERGE INTO {TGT} t USING changes s ON t.id = s.id
        WHEN MATCHED AND s.chg = 'D' THEN DELETE
        WHEN MATCHED THEN UPDATE SET t.v = s.v, t.op = 'upd'
        WHEN NOT MATCHED THEN INSERT (id, v, op) VALUES (s.id, s.v, 'ins')""")
for r in sql(f"SELECT op, count(*) AS n FROM {TGT} GROUP BY op ORDER BY op").collect():
    print(f"op={r.op:<5} rows={r.n}")
print("total rows =", sql(f"SELECT count(*) FROM {TGT}").collect()[0][0])
print("snapshot ops:", [r.operation for r in sql(f"SELECT operation FROM {TGT}.snapshots ORDER BY committed_at").collect()])
last = sql(f"""SELECT summary['added-records'] a, summary['deleted-records'] d
               FROM {TGT}.snapshots ORDER BY committed_at DESC LIMIT 1""").collect()[0]
print(f"merge snapshot: added-records={last.a} deleted-records={last.d}")

print("\n== 2. duplicate keys in the source ==")
# id 100 is an untouched row still in the target
dup_match = spark.createDataFrame([(100, "a", "U", 1), (100, "b", "U", 2)], "id BIGINT, v STRING, chg STRING, log_seq INT")
dup_match.createOrReplaceTempView("dup_changes")
try:
    sql(f"""MERGE INTO {TGT} t USING dup_changes s ON t.id = s.id
            WHEN MATCHED THEN UPDATE SET t.v = s.v
            WHEN NOT MATCHED THEN INSERT (id, v, op) VALUES (s.id, s.v, 'ins')""")
    print("duplicate keys matching a target row: MERGE succeeded")
except Exception as e:
    print("duplicate keys matching a target row:", err(e))
dup_new = spark.createDataFrame([(5000, "a", "I", 1), (5000, "b", "I", 2)], "id BIGINT, v STRING, chg STRING, log_seq INT")
dup_new.createOrReplaceTempView("dup_new")
sql(f"""MERGE INTO {TGT} t USING dup_new s ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET t.v = s.v
        WHEN NOT MATCHED THEN INSERT (id, v, op) VALUES (s.id, s.v, 'ins')""")
print("duplicate keys that match nothing: rows with id 5000 after MERGE =",
      sql(f"SELECT count(*) FROM {TGT} WHERE id = 5000").collect()[0][0])
w = Window.partitionBy("id").orderBy(F.col("log_seq").desc())
latest = dup_match.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")
latest.createOrReplaceTempView("dedup_changes")
sql(f"""MERGE INTO {TGT} t USING dedup_changes s ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET t.v = s.v""")
print("after deduplicating by log_seq: id 100 has v =",
      sql(f"SELECT v FROM {TGT} WHERE id = 100").collect()[0][0])

print("\n== 3. a match condition that is not the key ==")
NK = f"{NS}.customers"
sql(f"DROP TABLE IF EXISTS {NK} PURGE")
sql(f"CREATE TABLE {NK} (id BIGINT, email STRING, tier STRING) USING iceberg")
spark.createDataFrame([(1, "a@x.com", "free"), (2, "a@x.com", "free"), (3, "b@x.com", "free")],
                      "id BIGINT, email STRING, tier STRING").writeTo(NK).append()
spark.createDataFrame([("a@x.com", "gold")], "email STRING, tier STRING").createOrReplaceTempView("tier_changes")
sql(f"MERGE INTO {NK} t USING tier_changes s ON t.email = s.email WHEN MATCHED THEN UPDATE SET t.tier = s.tier")
print("one source row, matched on a non-unique email: rows updated =",
      sql(f"SELECT count(*) FROM {NK} WHERE tier = 'gold'").collect()[0][0])

print("\n== 4. how many files a write produces ==")
spark.conf.set("spark.sql.adaptive.enabled", "false")   # show the raw task count
base = datetime(2026, 1, 1)
rows = spark.range(0, 60_000).selectExpr(
    "id", f"timestamp'{base:%Y-%m-%d} 00:00:00' + make_interval(0, 0, 0, 0, 0, 0, id * 43.2) AS event_ts",
    "element_at(array('us-east','us-west','eu-west','ap-south'), cast(id % 4 AS int) + 1) AS region",
    "cast(10 + (id % 97) * 1.5 AS decimal(10,2)) AS amount_usd").repartition(200)
print("input: 60,000 events over", rows.select(F.to_date("event_ts")).distinct().count(), "days,",
      rows.rdd.getNumPartitions(), "Spark partitions")
for mode in ["(default)", "none", "hash"]:
    t = f"{NS}.files_{mode.strip('()')}"
    sql(f"DROP TABLE IF EXISTS {t} PURGE")
    props = "" if mode == "(default)" else f"TBLPROPERTIES ('write.distribution-mode'='{mode}')"
    sql(f"""CREATE TABLE {t} (id BIGINT, event_ts TIMESTAMP, region STRING, amount_usd DECIMAL(10,2))
            USING iceberg PARTITIONED BY (days(event_ts)) {props}""")
    rows.writeTo(t).append()
    files = sql(f"SELECT count(*) FROM {t}.files").collect()[0][0]
    parts = sql(f"SELECT count(DISTINCT partition) FROM {t}.files").collect()[0][0]
    print(f"write.distribution-mode {mode:<10} -> {files:>5} files in {parts} partitions")

RP = f"{NS}.files_repartition"
sql(f"DROP TABLE IF EXISTS {RP} PURGE")
sql(f"""CREATE TABLE {RP} (id BIGINT, event_ts TIMESTAMP, region STRING, amount_usd DECIMAL(10,2))
        USING iceberg PARTITIONED BY (days(event_ts)) TBLPROPERTIES ('write.distribution-mode'='none')""")
(rows.repartition(24, "event_ts").sortWithinPartitions("region").writeTo(RP).append())
print("repartition(24, 'event_ts'), mode none      ->",
      sql(f"SELECT count(*) FROM {RP}.files").collect()[0][0], "files")
sql(f"DELETE FROM {RP}")
(rows.repartition(24, F.to_date("event_ts")).sortWithinPartitions("region").writeTo(RP).append())
print("repartition(24, to_date(event_ts)), mode none ->",
      sql(f"SELECT count(*) FROM {RP}.files WHERE record_count > 0").collect()[0][0], "files")
spark.conf.set("spark.sql.adaptive.enabled", "true")

sql(f"""SELECT partition, count(*) AS files,
               round(avg(file_size_in_bytes) / 1048576, 3) AS avg_mb,
               sum(record_count) AS rows
        FROM {NS}.files_none.files
        GROUP BY partition ORDER BY files DESC LIMIT 3""").show(truncate=False)

print("== 5. writing by path instead of through the catalog ==")
BP = f"{NS}.by_path"
sql(f"DROP TABLE IF EXISTS {BP} PURGE")
sql(f"CREATE TABLE {BP} (id BIGINT, v STRING) USING iceberg")
spark.createDataFrame([(1, "a")], "id BIGINT, v STRING").writeTo(BP).append()
loc = sql(f"DESCRIBE TABLE EXTENDED {BP}").where("col_name = 'Location'").collect()[0][1]
try:
    (spark.createDataFrame([(2, "b")], "id BIGINT, v STRING")
          .write.format("iceberg").mode("append").save(loc))
    print("save(path) succeeded")
except Exception as e:
    print("save(path) on a Hive-catalog table:", err(e))
sql(f"REFRESH TABLE {BP}")
print("rows the catalog sees:", sql(f"SELECT count(*) FROM {BP}").collect()[0][0])
spark.createDataFrame([(3, "c")], "id BIGINT, v STRING").write.format("iceberg").mode("append").save(BP)
print("save('ice.blog_ingestion.by_path') by name: rows =", sql(f"SELECT count(*) FROM {BP}").collect()[0][0])

print("\n== 6. add_files: adopting Parquet files in place ==")
AF = f"{NS}.adopted"
sql(f"DROP TABLE IF EXISTS {AF} PURGE")
sql(f"CREATE TABLE {AF} (id BIGINT, event_ts TIMESTAMP, region STRING, amount_usd DECIMAL(10,2)) USING iceberg")
rows.limit(500).coalesce(2).write.mode("overwrite").parquet(f"{SRC}/events_ok")
sql(f"CALL ice.system.add_files(table => 'blog_ingestion.adopted', source_table => '`parquet`.`{SRC}/events_ok`')").show()
print("rows after add_files:", sql(f"SELECT count(*) FROM {AF}").collect()[0][0],
      "| data files point at:", sql(f"SELECT file_path FROM {AF}.files LIMIT 1").collect()[0][0].split("/blog_ingestion_src/")[0] + "/blog_ingestion_src/...")
(rows.limit(100).selectExpr("id", "cast(event_ts AS string) AS event_ts", "region")
     .coalesce(1).write.mode("overwrite").parquet(f"{SRC}/events_bad"))
try:
    sql(f"CALL ice.system.add_files(table => 'blog_ingestion.adopted', source_table => '`parquet`.`{SRC}/events_bad`')").show()
    print("add_files with a mismatched schema succeeded; rows =", sql(f"SELECT count(*) FROM {AF}").collect()[0][0])
    try:
        sql(f"SELECT count(*), min(event_ts) FROM {AF}").show()
    except Exception as e:
        print("reading the table afterwards:", err(e))
except Exception as e:
    print("add_files with a mismatched schema:", err(e))
try:
    sql(f"CALL ice.system.add_files(table => 'blog_ingestion.adopted', source_table => '`parquet`.`{SRC}/events_ok`')")
    print("adding the same files twice succeeded")
except Exception as e:
    print("adding the same files twice:", err(e))


def parquet_files(path):
    jpath = spark._jvm.org.apache.hadoop.fs.Path(path)
    fs = jpath.getFileSystem(spark._jsc.hadoopConfiguration())
    if not fs.exists(jpath):
        return 0
    it = fs.listFiles(jpath, True)
    n = 0
    while it.hasNext():
        n += it.next().getPath().getName().endswith(".parquet")
    return n


print("source Parquet files before DROP TABLE ... PURGE:", parquet_files(f"{SRC}/events_ok"))
sql(f"DROP TABLE {AF} PURGE")
print("source Parquet files after  DROP TABLE ... PURGE:", parquet_files(f"{SRC}/events_ok"))

print("\n== 7. CREATE TABLE AS SELECT with properties ==")
SUM = f"{NS}.events_summary"
sql(f"DROP TABLE IF EXISTS {SUM} PURGE")
sql(f"""CREATE TABLE {SUM}
        USING iceberg
        PARTITIONED BY (days(event_ts))
        TBLPROPERTIES ('write.target-file-size-bytes' = '268435456')
        AS SELECT event_ts, region, count(*) AS n, sum(amount_usd) AS total
           FROM {NS}.files_hash GROUP BY event_ts, region""")
print("CTAS: rows =", sql(f"SELECT count(*) FROM {SUM}").collect()[0][0],
      "| files =", sql(f"SELECT count(*) FROM {SUM}.files").collect()[0][0],
      "| target-file-size-bytes =",
      sql(f"SHOW TBLPROPERTIES {SUM} ('write.target-file-size-bytes')").collect()[0][1])

print("\n== 8. guarding against an empty load ==")
daily_df = rows.where("to_date(event_ts) = DATE'2026-02-15'")
try:
    if daily_df.head(1):
        daily_df.writeTo(f"{NS}.files_hash").overwritePartitions()
    else:
        raise ValueError("refusing to run a load that produced no rows")
except ValueError as e:
    print("guard:", e)
spark.stop()