Iceberg with Spark

Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to compute engines, including Spark, Trino, PrestoDB, Flink, Hive, and Impala, using a high-performance table format that works just like an SQL table.

User Experience

Iceberg avoids unpleasant surprises. Schema evolution works and won't inadvertently un-delete data. Users don't need to know about partitioning to get fast queries.

• Schema evolution: Supports adding, dropping, updating, or renaming operations without unintended consequences.
• Hidden partitioning: Prevents user errors that can lead to silently incorrect results or dramatically slow queries.
• Partition layout evolution: Updates the layout of a table as data volume or query patterns change.
• Time travel: Enables reproducible queries that use the same table snapshot or lets users easily examine changes.
• Version Rollback: Allows users to quickly correct problems by resetting tables to a good state.

Spark/Iceberg Compatibility Matrix

Choose the Iceberg Runtime for Your Spark Version

Use the Iceberg runtime JAR that matches your Spark version. The Iceberg runtime is compiled against Spark-specific analyzer and extension classes. Using a runtime built for a different Spark version can cause extension-loading errors, such as AbstractMethodError. For more information, see Troubleshooting.

For cloud storage with Spark 4.1.1 and other Spark3 versions, also add the required Iceberg bundle:

• AWS: iceberg-aws-bundle:1.11.0
• Azure: iceberg-azure-bundle:1.11.0
• Google Cloud: iceberg-gcp-bundle:1.11.0

Important considerations

• Spark 4.1.1 uses Scala 2.13. Use the _2.13 Iceberg runtime. Do not use a _2.12 runtime with Spark 4.1.1.
• Do not use the Spark 3.5 Iceberg runtime with Spark 4.1.1. This mismatch can cause an AbstractMethodError, including failures related to RewriteUpdateTableForRowLineage. For more information, see Troubleshooting.
• Use Iceberg 1.8.1 with Spark 3.3.3 when running on JDK 8. Iceberg 1.9 and later require JDK 11-compatible bytecode and can fail on JDK 8.
• Upgrade legacy Iceberg 1.4.x runtimes before using them for production workloads. The released POM files for Spark 3.3.3 and Spark 3.5.1 pin iceberg.version=1.4.3, which is significantly older than the recommended runtimes in the table above. Until an updated Mpack ships, side-load the recommended runtime (iceberg-spark-runtime-3.3_2.12:1.8.1 for Spark 3.3.3, iceberg-spark-runtime-3.5_2.12:1.10.2 for Spark 3.5.1) via --jars. See Known Limitations Iceberg for details.

Configure Spark Shell and Spark SQL with the Hive Metastore Catalog

Recommended: Use the Hive metastore catalog for ODP 3.3.6.5 and later.

Use SparkSessionCatalog with the Hive metastore to access Iceberg and standard Hive tables through the same spark_catalog namespace.

Configure Spark Shell with the following options:

/usr/odp/current/spark3-client/bin/spark-shell \
  --master yarn --deploy-mode client \
  --jars /path/to/iceberg-spark-runtime-3.5_2.12-1.10.2.jar \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
  --conf spark.sql.catalog.spark_catalog.type=hive \
  --conf spark.sql.warehouse.dir=hdfs:///apps/hive/warehouse

Replace spark3-client with spark4-client and the runtime jar with iceberg-spark-runtime-4.1_2.13-1.11.0.jar for Spark 4.1.1. Use the per-line paths and jars from the previous section.

Iceberg tables created with this configuration are registered in the Hive metastore and are available in Spark as:

spark_catalog.<database>.<table>

Access Iceberg metadata tables

When you use SparkSessionCatalog, specify the fully qualified name when accessing Iceberg metadata tables.

For example:

SELECT count(*)
FROM spark_catalog.default.sales.snapshots;

Do not use the short table name:

SELECT count(*)
FROM sales.snapshots;

Using the short form can result in a Table or view not found error.

Hive Metastore Compatibility

In releases earlier than ODP 3.3.6.5, using SparkSessionCatalog with type=hive can fail during the first Hive Metastore call with the following error:

org.apache.thrift.TApplicationException: Invalid method name: 'get_table'

This issue occurs because Hive 4 removed legacy Hive Metastore APIs, including get_table and get_table_objects_by_name, that are used by Spark's Hive 2.3 compatibility layer.

ODP 3.3.6.5 and later include a compatibility fix that restores the required APIs in the Hive 4 Metastore. No metastore version overrides or workaround configurations are required when you use SparkSessionCatalog with type=hive.

If this error occurs on ODP 3.3.6.5 or later, restart the Hive service from Ambari after installing the Mpack.

Configure Spark with HadoopCatalog

If the Hive Metastore (HMS) is unavailable or not required, use HadoopCatalog. This configuration is independent of HMS and is useful for smoke tests or environments that don't use HMS.

--conf spark.sql.catalog.hd=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.hd.type=hadoop \
--conf spark.sql.catalog.hd.warehouse=hdfs:///tmp/iceberg_warehouse

Tables created with this configuration are available under:

hd.<database>.<table>

Cross-Spark-version interop

Iceberg tables in ODP 3.3.6.5 round-trip cleanly across every Spark line on a shared HMS + HDFS location, provided each Spark job loads the matching runtime for its Spark version:

Iceberg's table format is the compatibility contract — the Spark version is not. Users do not need to standardise on a single Spark line to write and read the same Iceberg lake from multiple engines.

Import Statements

Include the necessary Apache Iceberg and Spark libraries to enable data operations.

import org.apache.spark.sql.types._
import org.apache.spark.sql.Row

Perform CRUD Operations on an Iceberg Table

After configuring the Iceberg catalog, use the following example to create an Iceberg table and perform insert, update, delete, and merge operations. The example applies to all supported Spark versions.

Create an Iceberg table

Create a partitioned Iceberg table in the default database.

CREATE TABLE spark_catalog.default.sales (
    id BIGINT,
    name STRING,
    region STRING,
    amount DOUBLE
)
USING iceberg
PARTITIONED BY (region);

Insert data

Insert sample records into the table.

INSERT INTO spark_catalog.default.sales VALUES
    (1, 'alice', 'us', 10.0),
    (2, 'bob', 'eu', 20.0),
    (3, 'carol', 'ap', 30.0),
    (4, 'dave', 'us', 40.0);

Update data

Update the amount value for a specific record.

UPDATE spark_catalog.default.sales
SET amount = 99.9
WHERE id = 2;

Delete data

Delete records for a specific region.

DELETE FROM spark_catalog.default.sales
WHERE region = 'ap';

Merge data

Use MERGE INTO to update matching records or insert new records.

MERGE INTO spark_catalog.default.sales AS target
USING (
    SELECT
        5 AS id,
        'eve' AS name,
        'eu' AS region,
        50.0 AS amount
) AS source
ON target.id = source.id
WHEN MATCHED THEN
    UPDATE SET target.amount = source.amount
WHEN NOT MATCHED THEN
    INSERT *;

View snapshots

Query the Iceberg metadata table to view the available snapshots.

SELECT count(*)
FROM spark_catalog.default.sales.snapshots;

Run a time travel query

Use a snapshot ID to query an earlier version of the table.

SELECT *
FROM spark_catalog.default.sales
VERSION AS OF <snapshot_id>;

For more details, see Apache Iceberg Getting Started Guide.

Configure Iceberg File Formats

Apache Iceberg supports Parquet, ORC, and Avro as storage file formats. Parquet is the default format. To use a different format, configure the write.format.default table property.\

Parquet

• Parquet is the default file format for Iceberg tables and does not require additional configuration.
• The CRUD examples in the previous section use Parquet by default.

ORC

To store an Iceberg table in ORC format, set the write.format.default table property to orc when you create the table.

CREATE TABLE spark_catalog.default.sales_orc (
    id BIGINT,
    name STRING,
    region STRING
)
USING iceberg
PARTITIONED BY (region)
TBLPROPERTIES ('write.format.default' = 'orc');

Insert data into the ORC-backed Iceberg table:

INSERT INTO spark_catalog.default.sales_orc VALUES
    (1, 'alice', 'us'),
    (2, 'bob', 'eu');

This configuration is validated with the recommended Iceberg runtimes for Spark 3.3.3, 3.5.1, 3.5.5, and 4.1.1.

Avro

To use Avro as the storage format, set the write.format.default table property to avro:

TBLPROPERTIES ('write.format.default' = 'avro');

Avro is supported by the Iceberg runtime but was not validated with ODP 3.3.6.5.

Troubleshooting

Use the following information to troubleshoot common Apache Iceberg issues with Spark.

Resolve "Table or view not found" for Metadata Tables

When you use SparkSessionCatalog, accessing an Iceberg metadata table by its short name can result in a Table or view not found error.

For example, do not use:

SELECT count(*)
FROM sales.snapshots;

Instead, specify the fully qualified catalog, database, and table name:

SELECT count(*)
FROM spark_catalog.default.sales.snapshots;

When you use SparkSessionCatalog, specify metadata tables as:

spark_catalog.<database>.<table>.<metadata_table>

For example:

spark_catalog.default.sales.snapshots

SparkCatalog configured with type=hadoop accepts the short form, while SparkSessionCatalog requires the fully qualified name.

Resolve "Invalid method name: 'get_table'"

In releases earlier than ODP 3.3.6.5, creating an Iceberg table using SparkSessionCatalog with type=hive can fail with the following error:

org.apache.thrift.TApplicationException: Invalid method name: 'get_table'

ODP 3.3.6.5 and later include the Hive Metastore compatibility fix required by Spark.

If this error occurs on ODP 3.3.6.5 or later, restart HiveServer2 from Ambari and retry the operation.

For more information, see Hive Metastore Compatibility.