KnowForge/Apache Iceberg 1.11.0/ 返回书籍
Apache Spark

Configuration

qianmoQqianmoQ· 更新于 2026-09-21· 阅读 46 分钟· 0 次阅读

登录后可跨设备保存划线和私人笔记登录

Spark Configuration🔗

Catalogs🔗

Spark adds an API to plug in table catalogs that are used to load, create, and manage Iceberg tables. Spark catalogs are configured by setting Spark properties under spark.sql.catalog.

This creates an Iceberg catalog named hive_prod that loads tables from a Hive metastore:

spark.sql.catalog.hive_prod = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.hive_prod.type = hive
spark.sql.catalog.hive_prod.uri = thrift://metastore-host:port
# omit uri to use the same URI as Spark: hive.metastore.uris in hive-site.xml

Below is an example for a REST catalog named rest_prod that loads tables from REST URL http://localhost:8080:

spark.sql.catalog.rest_prod = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.rest_prod.type = rest
spark.sql.catalog.rest_prod.uri = http://localhost:8080

Iceberg also supports a directory-based catalog in HDFS that can be configured using type=hadoop:

spark.sql.catalog.hadoop_prod = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.hadoop_prod.type = hadoop
spark.sql.catalog.hadoop_prod.warehouse = hdfs://nn:8020/warehouse/path

Info

The Hive-based catalog only loads Iceberg tables. To load non-Iceberg tables in the same Hive metastore, use a session catalog.

Catalog configuration🔗

A catalog is created and named by adding a property spark.sql.catalog.(catalog-name) with an implementation class for its value.

Iceberg supplies two implementations:

  • org.apache.iceberg.spark.SparkCatalog supports a Hive Metastore or a Hadoop warehouse as a catalog
  • org.apache.iceberg.spark.SparkSessionCatalog adds support for Iceberg tables to Spark's built-in catalog, and delegates to the built-in catalog for non-Iceberg tables

Both catalogs are configured using properties nested under the catalog name. Common configuration properties for Hive and Hadoop are:

PropertyValuesDescription
spark.sql.catalog.catalog-name.typehive, hadoop, rest, glue, jdbc or nessieThe underlying Iceberg catalog implementation, HiveCatalog, HadoopCatalog, RESTCatalog, GlueCatalog, JdbcCatalog, NessieCatalog or left unset if using a custom catalog
spark.sql.catalog.catalog-name.catalog-implThe custom Iceberg catalog implementation. If type is null, catalog-impl must not be null.
spark.sql.catalog.catalog-name.io-implThe custom FileIO implementation.
spark.sql.catalog.catalog-name.metrics-reporter-implThe custom MetricsReporter implementation.
spark.sql.catalog.catalog-name.default-namespacedefaultThe default current namespace for the catalog
spark.sql.catalog.catalog-name.urithrift://host:portHive metastore URL for hive typed catalog, REST URL for REST typed catalog
spark.sql.catalog.catalog-name.warehousehdfs://nn:8020/warehouse/pathBase path for the warehouse directory
spark.sql.catalog.catalog-name.cache-enabledtrue or falseWhether to enable catalog cache, default value is true
spark.sql.catalog.catalog-name.cache.expiration-interval-ms30000 (30 seconds)Duration after which cached catalog entries are expired; Only effective if cache-enabled is true. -1 disables cache expiration and 0 disables caching entirely, irrespective of cache-enabled. Default is 30000 (30 seconds)
spark.sql.catalog.catalog-name.table-default.propertyKeyDefault Iceberg table property value for property key propertyKey, which will be set on tables created by this catalog if not overridden
spark.sql.catalog.catalog-name.table-override.propertyKeyEnforced Iceberg table property value for property key propertyKey, which cannot be overridden on table creation by user
spark.sql.catalog.catalog-name.view-default.propertyKeyDefault Iceberg view property value for property key propertyKey, which will be set on views created by this catalog if not overridden
spark.sql.catalog.catalog-name.view-override.propertyKeyEnforced Iceberg view property value for property key propertyKey, which cannot be overridden on view creation by user
spark.sql.catalog.catalog-name.use-nullable-query-schematrue or falseWhether to preserve fields' nullability when creating the table using CTAS and RTAS. If set to true, all fields will be marked as nullable. If set to false, fields' nullability will be preserved. The default value is true. Available in Spark 3.5 and above.

Additional properties can be found in common catalog configuration.

Using catalogs🔗

Catalog names are used in SQL queries to identify a table. In the examples above, hive_prod and hadoop_prod can be used to prefix database and table names that will be loaded from those catalogs.

SELECT * FROM hive_prod.db.table; -- load db.table from catalog hive_prod

Spark 3 keeps track of the current catalog and namespace, which can be omitted from table names.

USE hive_prod.db;
SELECT * FROM table; -- load db.table from catalog hive_prod

To see the current catalog and namespace, run SHOW CURRENT NAMESPACE.

Replacing the session catalog🔗

To add Iceberg table support to Spark's built-in catalog, configure spark_catalog to use Iceberg's SparkSessionCatalog.

spark.sql.catalog.spark_catalog = org.apache.iceberg.spark.SparkSessionCatalog
spark.sql.catalog.spark_catalog.type = hive

Spark's built-in catalog supports existing v1 and v2 tables tracked in a Hive Metastore. This configures Spark to use Iceberg's SparkSessionCatalog as a wrapper around that session catalog. When a table is not an Iceberg table, the built-in catalog will be used to load it instead.

This configuration can use same Hive Metastore for both Iceberg and non-Iceberg tables.

SparkSessionCatalog is useful when you want spark_catalog to work with both Iceberg and non-Iceberg tables in the same metastore.

Note

Spark before 4.2.0 does not support V2Function in the session catalog. See SPARK-54760 (apache/spark#53531) for details. As a result, catalog-scoped SQL functions such as system.bucket, system.days, and system.iceberg_version are not available through spark_catalog. To work around this limitation, configure a separate Iceberg catalog with org.apache.iceberg.spark.SparkCatalog and call them through that catalog.

Using catalog specific Hadoop configuration values🔗

Similar to configuring Hadoop properties by using spark.hadoop.*, it's possible to set per-catalog Hadoop configuration values when using Spark by adding the property for the catalog with the prefix spark.sql.catalog.(catalog-name).hadoop.*. These properties will take precedence over values configured globally using spark.hadoop.* and will only affect Iceberg tables.

spark.sql.catalog.hadoop_prod.hadoop.fs.s3a.endpoint = http://aws-local:9000

Loading a custom catalog🔗

Spark supports loading a custom Iceberg Catalog implementation by specifying the catalog-impl property. Here is an example:

spark.sql.catalog.custom_prod = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.custom_prod.catalog-impl = com.my.custom.CatalogImpl
spark.sql.catalog.custom_prod.my-additional-catalog-config = my-value

SQL Extensions🔗

Iceberg 0.11.0 and later add an extension module to Spark to add new SQL commands, like CALL for stored procedures or ALTER TABLE ... WRITE ORDERED BY.

Using those SQL commands requires adding Iceberg extensions to your Spark environment using the following Spark property:

Spark extensions propertyIceberg extensions implementation
spark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Runtime configuration🔗

Precedence of Configuration Settings🔗

Iceberg allows configurations to be specified at different levels. The effective configuration for a read or write operation is determined based on the following order of precedence:

  1. DataSource API Read/Write Options – Explicitly passed to .option(...) in a read/write operation.
  2. Spark Session Configuration - Set globally in Spark via spark.conf.set(...), spark-defaults.conf, or --conf in spark-submit.
  3. Table Properties – Defined on the Iceberg table via ALTER TABLE SET TBLPROPERTIES.
  4. Default Value.

If a setting is not defined at a higher level, the next level is used as fallback. This allows flexibility while enabling global defaults when needed.

Spark SQL Options🔗

Iceberg supports setting various global behaviors using Spark SQL configuration options. These can be set via spark.conf, SparkSession settings, or Spark submit arguments. For example:

// disabling vectorization
val spark = SparkSession.builder()
  .appName("IcebergExample")
  .master("local[*]")
  .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog")
  .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
  .config("spark.sql.iceberg.vectorization.enabled", "false")
  .getOrCreate()
Spark optionDefaultDescription
spark.sql.iceberg.vectorization.enabledTable defaultEnables vectorized reads of data files
spark.sql.iceberg.check-nullabilitytrueValidate that the write schema's nullability matches the table's nullability
spark.sql.iceberg.check-orderingtrueValidates the write schema column order matches the table schema order
spark.sql.iceberg.planning.preserve-data-groupingfalseWhen true, co-locate scan tasks for the same partition in the same read split, used in Storage Partitioned Joins
spark.sql.iceberg.aggregate-push-down.enabledtrueEnables pushdown of aggregate functions (MAX, MIN, COUNT)
spark.sql.iceberg.distribution-modeSee Spark WritesControls distribution strategy during writes
spark.wap.idnullWrite-Audit-Publish snapshot staging ID
spark.wap.branchnullWAP branch name for snapshot commit
spark.sql.iceberg.shred-variantsTable defaultWhen true, variant columns are written with shredded Parquet encoding for improved query performance
spark.sql.iceberg.variant-inference-buffer-sizeTable defaultNumber of rows to buffer for schema inference when variant shredding is enabled
spark.sql.iceberg.compression-codecTable defaultWrite compression codec (e.g., zstd, snappy)
spark.sql.iceberg.compression-levelTable defaultCompression level for Parquet/Avro
spark.sql.iceberg.compression-strategyTable defaultCompression strategy for ORC
spark.sql.iceberg.data-planning-modeAUTOScan planning mode for data files (AUTO, LOCAL, DISTRIBUTED)
spark.sql.iceberg.delete-planning-modeAUTOScan planning mode for delete files (AUTO, LOCAL, DISTRIBUTED)
spark.sql.iceberg.advisory-partition-sizeTable defaultAdvisory size (bytes) used for writing to the Table when Spark's Adaptive Query Execution is enabled. Used to size output files
spark.sql.iceberg.locality.enabledfalseReport locality information for Spark task placement on executors
spark.sql.iceberg.executor-cache.enabledtrueEnables cache for executor-side (currently used to cache Delete Files)
spark.sql.iceberg.executor-cache.timeout10Timeout in minutes for executor cache entries
spark.sql.iceberg.executor-cache.max-entry-size67108864 (64MB)Max size per cache entry (bytes)
spark.sql.iceberg.executor-cache.max-total-size134217728 (128MB)Max total executor cache size (bytes)
spark.sql.iceberg.executor-cache.locality.enabledfalseEnables locality-aware executor cache usage
spark.sql.iceberg.merge-schemafalseEnables modifying the table schema to match the write schema. Only adds columns missing columns
spark.sql.iceberg.report-column-statstrueReport Puffin Table Statistics if available to Spark's Cost Based Optimizer. CBO must be enabled for this to be effective
spark.sql.iceberg.async-micro-batch-planning-enabledfalseEnables asynchronous microbatch planning to reduce planning latency by pre-fetching file scan tasks

Read options🔗

Spark read options are passed when configuring the DataFrameReader, like this:

// time travel
spark.read
    .option("snapshot-id", 10963874102873L)
    .table("catalog.db.table")
Spark optionDefaultDescription
snapshot-id(latest)Snapshot ID of the table snapshot to read
as-of-timestamp(latest)A timestamp in milliseconds; the snapshot used will be the snapshot current at this time.
split-sizeAs per table propertyOverrides this table's read.split.target-size and read.split.metadata-target-size
lookbackAs per table propertyOverrides this table's read.split.planning-lookback
file-open-costAs per table propertyOverrides this table's read.split.open-file-cost
vectorization-enabledAs per table propertyOverrides this table's read.parquet.vectorization.enabled
batch-sizeAs per table propertyOverrides this table's read.parquet.vectorization.batch-size
stream-from-timestamp(none)A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used
streaming-max-files-per-micro-batchINT_MAXMaximum number of files per microbatch
streaming-max-rows-per-micro-batchINT_MAX"Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit
async-micro-batch-planning-enabledfalseEnables asynchronous microbatch planning to reduce planning latency by pre-fetching file scan tasks
streaming-snapshot-polling-interval-ms30000Overrides the polling time for async planner to refresh and detect new snapshots. Only affects when async-micro-batch-planning-enabled is set
async-queue-preload-file-limit100Overrides the number of files loaded to background queue initially. Tune to prevent queue starvation. Only affects when async-micro-batch-planning-enabled is set
async-queue-preload-row-limit100000Overrides the number of rows loaded to background queue initially. Tune to prevent queue starvation. Only affects when async-micro-batch-planning-enabled is set

Write options🔗

Spark write options are passed when configuring the DataFrameWriterV2, like this:

// write with Avro instead of Parquet
df.writeTo("catalog.db.table")
    .option("write-format", "avro")
    .option("snapshot-property.key", "value")
    .append()
Spark optionDefaultDescription
write-formatTable write.format.defaultFile format to use for this write operation; parquet, avro, or orc
target-file-size-bytesAs per table propertyOverrides this table's write.target-file-size-bytes
check-nullabilitytrueSets the nullable check on fields
snapshot-property.custom-keynullAdds an entry with custom-key and corresponding value in the snapshot summary (the snapshot-property. prefix is only required for DSv2)
fanout-enabledfalseOverrides this table's write.spark.fanout.enabled
check-orderingtrueChecks if input schema and table schema are same
isolation-levelnullDesired isolation level for Dataframe overwrite operations. null => no checks (for idempotent writes), serializable => check for concurrent inserts or deletes in destination partitions, snapshot => checks for concurrent deletes in destination partitions.
validate-from-snapshot-idnullIf isolation level is set, id of base snapshot from which to check concurrent write conflicts into a table. Should be the snapshot before any reads from the table. Can be obtained via Table API or Snapshots table. If null, the table's oldest known snapshot is used.
compression-codecTable write.(fileformat).compression-codecOverrides this table's compression codec for this write
compression-levelTable write.(fileformat).compression-levelOverrides this table's compression level for Parquet and Avro tables for this write
compression-strategyTable write.orc.compression-strategyOverrides this table's compression strategy for ORC tables for this write
distribution-modeSee Spark Writes for defaultsOverride this table's distribution mode for this write
delete-granularityfileOverride this table's delete granularity for this write
shred-variantsfalseOverrides this table's write.parquet.shred-variants for this write
variant-inference-buffer-size100Overrides this table's write.parquet.variant-inference-buffer-size for this write

CommitMetadata provides an interface to add custom metadata to a snapshot summary during a SQL execution, which can be beneficial for purposes such as auditing or change tracking. If properties start with snapshot-property., then that prefix will be removed from each property. Here is an example:

import org.apache.iceberg.spark.CommitMetadata;

Map<String, String> properties = Maps.newHashMap();
properties.put("property_key", "property_value");
CommitMetadata.withCommitProperties(properties,
        () -> {
            spark.sql("DELETE FROM " + tableName + " where id = 1");
            return 0;
        },
        RuntimeException.class);

评论

登录后参与评论

正在加载评论…