Spark 配置

Catalog

Spark 提供了一套 API，可用于插入用于加载、创建和管理 Iceberg 表的 table catalog。Spark catalog 通过设置 spark.sql.catalog 下的 Spark 属性进行配置。

下面的配置会创建一个名为 hive_prod 的 Iceberg catalog，用于从 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

下面是针对名为 rest_prod 的 REST 目录的示例，该目录从 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 还支持 HDFS 中基于目录的 catalog，可通过 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

信息

基于 Hive 的目录只会加载 Iceberg 表。要在同一个 Hive 元存储中加载非 Iceberg 表，请使用会话目录。

目录配置

通过添加属性 spark.sql.catalog.(catalog-name) 并将其值设置为实现类，即可创建并命名一个目录。

Iceberg 提供两种实现：

• org.apache.iceberg.spark.SparkCatalog 支持将 Hive Metastore 或 Hadoop 数据仓库作为目录
• org.apache.iceberg.spark.SparkSessionCatalog 在 Spark 内置目录的基础上增加对 Iceberg 表的支持，并将非 Iceberg 表的请求委托给内置目录

两种目录都通过嵌套在目录名称下的属性进行配置。Hive 和 Hadoop 的常用配置属性如下：

其他属性可参见通用的目录配置。

使用目录

目录名称用于在 SQL 查询中标识表。在上面的示例中，hive_prod 和 hadoop_prod 可用作前缀，与数据库名和表名组合，从而从相应的目录中加载这些表。

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

Spark 3 会记录当前的 catalog 和命名空间，它们在表名中可以省略。

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

要查看当前目录和命名空间，请运行 SHOW CURRENT NAMESPACE。

替换会话目录

要为 Spark 内置目录添加 Iceberg 表支持，请将 spark_catalog 配置为使用 Iceberg 的 SparkSessionCatalog。

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

Spark 的内置 catalog 支持 Hive Metastore 中已有的 v1 和 v2 表。该配置使 Spark 使用 Iceberg 的 SparkSessionCatalog 作为该会话 catalog 的封装。当某个表不是 Iceberg 表时，将改用内置 catalog 来加载它。

通过此配置，Iceberg 表和非 Iceberg 表可以共用同一个 Hive Metastore。

当你希望 spark_catalog 在同一个 metastore 中同时处理 Iceberg 表和非 Iceberg 表时，SparkSessionCatalog 非常有用。

注意

Spark 4.2.0 之前的版本不支持会话 catalog 中的 V2Function。详情参见 SPARK-54760（apache/spark#53531）。因此，像 system.bucket、system.days 和 system.iceberg_version 这类基于 catalog 的 SQL 函数无法通过 spark_catalog 使用。要绕过此限制，请配置一个使用 org.apache.iceberg.spark.SparkCatalog 的独立 Iceberg catalog，并通过该 catalog 调用这些函数。

使用特定 catalog 的 Hadoop 配置项

与通过 spark.hadoop.* 配置 Hadoop 属性类似，在使用 Spark 时，可以通过添加带有前缀 spark.sql.catalog.(catalog-name).hadoop.* 的属性来设置每个 catalog 独立的 Hadoop 配置项。这些属性的优先级高于通过 spark.hadoop.* 全局配置的值，并且只会影响 Iceberg 表。

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

加载自定义目录

Spark 支持通过指定 catalog-impl 属性来加载自定义的 Iceberg Catalog 实现。示例如下：

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 扩展

Iceberg 0.11.0 及更高版本为 Spark 添加了一个扩展模块，用于引入新的 SQL 命令，例如用于存储过程的 CALL 以及 ALTER TABLE ... WRITE ORDERED BY。

使用这些 SQL 命令需要通过以下 Spark 属性将 Iceberg 扩展添加到你的 Spark 环境中：

运行时配置

配置项的优先级

Iceberg 允许在不同层级指定配置。读写操作的生效配置依据以下优先级顺序确定：

1. DataSource API 读/写选项 – 在读写操作中通过 .option(...) 显式传递。
2. Spark Session 配置 – 通过 spark.conf.set(...)、spark-defaults.conf 或 spark-submit 中的 --conf 在 Spark 中全局设置。
3. 表属性 – 通过 ALTER TABLE SET TBLPROPERTIES 定义在 Iceberg 表上。
4. 默认值。

如果某个设置在更高层级未定义，则会回退到下一层级。这样既保持了灵活性，又能在需要时启用全局默认值。

Spark SQL 选项

Iceberg 支持使用 Spark SQL 配置选项来设置各种全局行为。这些选项可以通过 spark.conf、SparkSession 设置或 Spark 提交参数进行配置。例如：

// 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 读取选项在配置 DataFrameReader 时传入，如下所示：

// time travel
spark.read
    .option("snapshot-id", 10963874102873L)
    .table("catalog.db.table")

写入选项

Spark 写入选项在配置 DataFrameWriterV2 时传入，示例如下：

// write with Avro instead of Parquet
df.writeTo("catalog.db.table")
    .option("write-format", "avro")
    .option("snapshot-property.key", "value")
    .append()

CommitMetadata 提供了一个接口，可在 SQL 执行过程中向快照摘要（snapshot summary）中添加自定义元数据，这对于审计或变更跟踪等用途非常有价值。如果属性名以 snapshot-property. 开头，则会从每个属性中去除该前缀。示例如下：

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);