Spark中使用ALTER TABLE CREATE BRANCH创建Iceberg分支遇解析错误

我通过以下代码创建Spark Connect客户端模式会话：

def get_spark_session(master_url: str) -> SparkSession:
    return (
        SparkSession.builder.remote(master_url)
        .config(
            "spark.sql.catalog.polaris",
            "org.apache.iceberg.spark.SparkCatalog",
        )
        .config(
            "spark.sql.catalog.polaris.type",
            "rest",
        )
        .config(
            "spark.sql.catalog.polaris.uri",
            os.getenv("CATALOG_URI"),
        )
        .config(
            "spark.sql.catalog.polaris.warehouse",
            os.getenv("CATALOG_NAME"),
        )
        .config("spark.sql.defaultCatalog", "polaris")
        .config(
            "spark.sql.catalog.polaris.scope",
            "PRINCIPAL_ROLE:ALL",
        )
        .config(
            "spark.sql.catalog.polaris.credential",
            f"{os.getenv("CATALOG_CLIENT_ID")}:{os.getenv("CATALOG_CLIENT_SECRET")}"
        )
        .config("spark.sql.catalog.polaris.client.region", "us-west-2")
        .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
 
        .getOrCreate()
    )

随后尝试为表创建分支：

if __name__ == "__main__":
    spark = get_spark_session(os.getenv("SPARK_MASTER_URL"))
    source_table = 'a.b.subscription_history'
    spark.sql(f"ALTER TABLE {source_table} CREATE BRANCH `dedup` RETAIN 7 DAYS WITH SNAPSHOT RETENTION 2 SNAPSHOTS")

执行后出现解析错误：

pyspark.errors.exceptions.connect.ParseException: 
[PARSE_SYNTAX_ERROR] Syntax error at or near 'CREATE'.(line 1, pos 46)`
== SQL ==
ALTER TABLE a.b.subscription_history CREATE BRANCH `dedup` RETAIN 7 DAYS WITH SNAPSHOT RETENTION 2 SNAPSHOTS
-------------------------------------^^^

我怀疑本地程序有问题，尝试在会话配置中添加jar包依赖：

.config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,org.apache.iceberg:iceberg-spark-extensions-3.5_2.12:1.5.2,org.apache.iceberg:iceberg-aws-bundle:1.5.2,org.apache.hadoop:hadoop-aws:3.3.4")

但没有效果，因为当前是Spark Connect客户端模式，jar包应该部署在Spark驱动服务器上。本地pip包列表如下：

Package                            Version
---------------------------------- -----------
annotated-types                    0.7.0
boto3                              1.35.57
botocore                           1.35.99
cachetools                         5.5.1
Deprecated                         1.2.18
googleapis-common-protos           1.70.0
grpcio                             1.73.1
grpcio-status                      1.73.1
importlib_metadata                 8.5.0
jmespath                           1.0.1
numpy                              1.26.4
opentelemetry-api                  1.30.0
opentelemetry-exporter-prometheus  0.51b0
opentelemetry-sdk                  1.30.0
opentelemetry-semantic-conventions 0.51b0
pandas                             2.3.0
pip                                24.2
prometheus_client                  0.21.0
protobuf                           6.31.1
psycopg2-binary                    2.9.10
py4j                               0.10.9.7
pyarrow                            20.0.0
pydantic                           2.9.2
pydantic_core                      2.23.4
pyspark                            3.5.2
python-dateutil                    2.9.0.post0
pytz                               2025.2
PyYAML                             6.0.2
s3transfer                         0.10.4
setuptools                         75.6.0
six                                1.17.0
tenacity                           9.0.0
typing_extensions                  4.14.0
tzdata                             2025.2
urllib3                            2.5.0
wrapt                              1.17.2
zipp                               3.23.0

请问该解析错误的原因是什么，如何解决？

这个解析错误的核心原因是Spark Connect客户端的SQL解析器未加载Iceberg的扩展语法。Spark Connect模式下，客户端会先对SQL语句做预解析，而本地客户端环境缺少Iceberg的SQL扩展支持，无法识别ALTER TABLE ... CREATE BRANCH这类Iceberg特有语法，直接抛出解析错误，请求根本没发送到服务器端。

你在客户端配置spark.jars.packages无效，是因为Spark Connect客户端仅负责发送请求，实际jar包依赖需部署在Spark驱动和集群节点上，但客户端的SQL预解析环节需要本地有对应语法支持才能通过校验。

1. 确保客户端环境加载Iceberg扩展 #

本地Python环境需确保PySpark能加载Iceberg的Spark扩展类，可通过以下方式处理：

• 启动Python程序前设置环境变量，添加Iceberg jar包依赖：

  export PYSPARK_SUBMIT_ARGS="--packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,org.apache.iceberg:iceberg-spark-extensions-3.5_2.12:1.5.2 pyspark-shell"
  

• 或在Python代码开头显式指定jar包：

  import pyspark
  pyspark.SparkContext.setSystemProperty("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,org.apache.iceberg:iceberg-spark-extensions-3.5_2.12:1.5.2")
  

2. 验证语法兼容性 #

确认使用的Iceberg 1.5.2版本支持CREATE BRANCH语法，可先简化语句验证：

ALTER TABLE a.b.subscription_history CREATE BRANCH `dedup`

若简化后能执行，再逐步添加保留策略参数。

3. 确认Spark服务器端的Iceberg配置 #

虽然客户端预解析是当前问题核心，但也要确保Spark驱动服务器端配置正确：

• 服务器端Spark配置已添加spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
• 服务器端部署了对应版本的Iceberg jar包
• 目录配置polaris在服务器端能正常访问

4. 使用Iceberg Python API替代SQL语句 #

若客户端SQL解析问题难以解决，可直接用Iceberg Python API创建分支，绕过客户端SQL预解析：

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = get_spark_session(os.getenv("SPARK_MASTER_URL"))
    # 获取Iceberg表对象
    table = spark.table('a.b.subscription_history')
    # 通过API创建分支
    table.create_branch("dedup", retain_days=7, snapshot_retention_count=2)

这种方式直接调用API，不会触发客户端语法校验，请求会直接发送到服务器端执行。

内容的提问来源于stack exchange，提问作者LuckyGambler