Spark 连接 Lakehouse（Iceberg REST）

概述

Lakehouse 提供了标准的 Apache Iceberg Catalog REST API 接口，允许外部计算引擎（如 Apache Spark）通过统一的 REST 协议访问和查询存储在 Lakehouse 中的 Iceberg 原生表。

核心特性

  • 标准兼容：兼容 Apache Iceberg REST Catalog 规范
  • 引擎支持：支持 Spark 3.5+ 计算引擎
  • 权限委托：通过 vended-credentials 模式管理存储访问权限
  • 多云支持：支持阿里云 OSS、AWS S3、腾讯云 COS 等

使用限制

表类型支持

表类型是否支持说明
Lakehouse 原生 Iceberg 表✅ 支持完整读写，数据正确性已验证
外部 Delta Lake 表❌ 不支持REST Catalog 接口返回 
Server error: Internal
Server error: Internal
外部 Hudi 表❌ 不支持同上
外部 Lance 表❌ 不支持同上
外部 Paimon 表❌ 不支持同上
Kafka 外部表❌ 不支持流式表无法通过 REST Catalog 扫描

数据类型限制

通过 Spark 访问时，以下数据类型暂不支持：

  • 整数类型：
    SMALLINT
    SMALLINT
    、
    TINYINT
    TINYINT
  • 半结构化类型：
    JSON
    JSON
  • 向量类型：
    VECTOR
    VECTOR

环境要求

组件版本要求备注
JDK17.x（必须）Java 18+ 移除了 
Subject.getSubject()
Subject.getSubject()
，导致 Hadoop 3.3 启动失败；Java 25/23 均无法使用
Apache Spark3.5+推荐 PySpark 3.5.5
Python3.x用于 PySpark 驱动
iceberg-spark-runtime1.6.1（Scala 2.12 / Spark 3.5.x）Maven：
org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1
org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1
aliyun-sdk-oss3.18.1阿里云 OSS 访问，Maven：
com.aliyun.oss:aliyun-sdk-oss:3.18.1
com.aliyun.oss:aliyun-sdk-oss:3.18.1

安装 PySpark

pip install "pyspark==3.5.5"

快速开始

连接参数说明

从 JDBC 连接串解析各配置字段：

jdbc:clickzetta://<instance_id>.<endpoint>.api.clickzetta.com/<workspace> ↑ instanceName ↑ region endpoint ↑ Workspace

REST Catalog URI 格式：

https://<endpoint>.api.clickzetta.com/api/v1/catalog/iceberg-rest
https://<endpoint>.api.clickzetta.com/api/v1/catalog/iceberg-rest
， 其他endpoint 取值请参考文档

PySpark 完整示例

import os import base64 from pyspark.sql import SparkSession # Spark 3.5 必须使用 JDK 17（Java 18+ 不兼容 Hadoop 3.3） os.environ["JAVA_HOME"] = os.path.expanduser("~/jdk17") os.environ["PATH"] = os.environ["JAVA_HOME"] + "/bin:" + os.environ.get("PATH", "") # 指定 Maven 仓库（本地 .m2 缓存不完整时可避免依赖下载失败） os.environ["PYSPARK_SUBMIT_ARGS"] = ( "--repositories https://repo1.maven.org/maven2 pyspark-shell" ) # ── 认证 ──────────────────────────────────────────────────── USERNAME = "your_username" PASSWORD = "your_password" auth_header = "Basic " + base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode() # ── SparkSession ───────────────────────────────────────────── spark = ( SparkSession.builder .appName("CZ-Iceberg") .config( "spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1," "com.aliyun.oss:aliyun-sdk-oss:3.18.1", ) .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") .config("spark.sql.defaultCatalog", "cz") .config("spark.sql.catalog.cz", "org.apache.iceberg.spark.SparkCatalog") .config("spark.sql.catalog.cz.type", "rest") .config("spark.sql.catalog.cz.uri", "https://<endpoint>.api.clickzetta.com/api/v1/catalog/iceberg-rest") .config("spark.sql.catalog.cz.header.instanceName", "your_instance_id") .config("spark.sql.catalog.cz.header.Workspace", "your_workspace") .config("spark.sql.catalog.cz.header.Authorization", auth_header) .config("spark.sql.catalog.cz.header.X-Iceberg-Access-Delegation", "vended-credentials") .config("spark.sql.catalog.cz.io-impl", "org.apache.iceberg.aliyun.oss.OSSFileIO") .config("spark.sql.catalog.cz.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com") .config("spark.sql.catalog.cz.default-namespace", "public") .getOrCreate() )

基本查询操作

# 查看所有 schema spark.sql("SHOW NAMESPACES IN cz").show() # 查看表列表 spark.sql("SHOW TABLES IN cz.public").show(100, truncate=False) # 查看表结构 spark.sql("DESCRIBE TABLE cz.public.your_table").show() # 查询数据 spark.sql("SELECT * FROM cz.public.your_table LIMIT 10").show() # DataFrame API df = spark.table("cz.public.your_table") df.filter("amount > 100").select("id", "amount").show()

配置参数详解

参数参数描述示例值是否必须
spark.jars.packages
spark.jars.packages
指定 Spark 启动时从 Maven 自动下载的依赖包，包含 Iceberg Spark runtime 和云对象存储 SDK
org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,com.aliyun.oss:aliyun-sdk-oss:3.18.1
org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,com.aliyun.oss:aliyun-sdk-oss:3.18.1
是
spark.sql.extensions
spark.sql.extensions
向 Spark SQL 注入 Iceberg 扩展，使其能解析 Iceberg 特有的 DDL/DML 语句
org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
是
spark.sql.defaultCatalog
spark.sql.defaultCatalog
设置 Spark SQL 的默认 Catalog，设置后 SQL 中可省略 catalog 前缀
cz
cz
否
spark.sql.catalog.cz
spark.sql.catalog.cz
注册名为 
cz
cz
的 Iceberg Catalog，指定实现类
org.apache.iceberg.spark.SparkCatalog
org.apache.iceberg.spark.SparkCatalog
是
spark.sql.catalog.cz.type
spark.sql.catalog.cz.type
指定 Catalog 类型为 REST，通过 HTTP 与远端元数据服务通信
rest
rest
是
spark.sql.catalog.cz.uri
spark.sql.catalog.cz.uri
REST Catalog 服务的 API 端点，Spark 向此地址发送所有元数据请求
https://cn-hangzhou-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
https://cn-hangzhou-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
是
spark.sql.catalog.cz.header.instanceName
spark.sql.catalog.cz.header.instanceName
HTTP 请求头，用于标识目标 Lakehouse 实例（JDBC URL 中主机名的第一段）
be6213f1
be6213f1
是
spark.sql.catalog.cz.header.Workspace
spark.sql.catalog.cz.header.Workspace
HTTP 请求头，指定要操作的 Workspace 名称
quick_start
quick_start
是
spark.sql.catalog.cz.header.Authorization
spark.sql.catalog.cz.header.Authorization
HTTP 认证头，值为 
Basic <base64(username:password)>
Basic <base64(username:password)>
Basic R19ZVU5RSTo=...
Basic R19ZVU5RSTo=...
是
spark.sql.catalog.cz.header.X-Iceberg-Access-Delegation
spark.sql.catalog.cz.header.X-Iceberg-Access-Delegation
启用 Vended Credentials 模式，由 Catalog 服务下发临时存储凭证，避免在客户端暴露长期密钥
vended-credentials
vended-credentials
是
spark.sql.catalog.cz.io-impl
spark.sql.catalog.cz.io-impl
指定读写数据文件（Parquet 等）的 FileIO 实现，阿里云使用 OSSFileIO
org.apache.iceberg.aliyun.oss.OSSFileIO
org.apache.iceberg.aliyun.oss.OSSFileIO
是
spark.sql.catalog.cz.oss.endpoint
spark.sql.catalog.cz.oss.endpoint
阿里云 OSS 区域端点，客户端通过此地址访问存储桶中的数据文件（可根据 OSS 所在区域修改，参考文档）
oss-cn-hangzhou.aliyuncs.com
oss-cn-hangzhou.aliyuncs.com
是
spark.sql.catalog.cz.default-namespace
spark.sql.catalog.cz.default-namespace
Catalog 内的默认命名空间（Schema），未显式指定时的默认操作目标
public
public
否（推荐）

区域端点对照

实例区域REST Catalog URIOSS Endpoint
阿里云杭州
https://cn-hangzhou-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
https://cn-hangzhou-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
oss-cn-hangzhou.aliyuncs.com
oss-cn-hangzhou.aliyuncs.com
阿里云上海
https://cn-shanghai-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
https://cn-shanghai-alicloud.api.clickzetta.com/api/v1/catalog/iceberg-rest
oss-cn-shanghai.aliyuncs.com
oss-cn-shanghai.aliyuncs.com

更多区域参考云器支持的云平台文档。

ServiceFailureException: Server error: Internal
ServiceFailureException: Server error: Internal

原因：外部格式表（Delta/Hudi/Lance/Paimon）不支持通过 Iceberg REST Catalog 访问，元数据可枚举但数据读取失败。

解决：此类表需通过 Lakehouse Studio 或 JDBC 直接查询，不支持 Spark Iceberg REST 路径。

BadRequestException: Unsupported path: .../metrics
BadRequestException: Unsupported path: .../metrics

原因：Iceberg 客户端在读取完成后尝试上报扫描指标，但 Lakehouse REST Catalog 暂不实现该接口（

/metrics
/metrics
路径）。

影响：仅是日志中的 WARN 级别错误，不影响数据读取正确性，可忽略。可关闭指标上报：

.config("spark.sql.catalog.cz.metrics-reporter-impl", "org.apache.iceberg.metrics.LoggingMetricsReporter")

认证失败（401）

  1. 检查用户名和密码是否正确
  2. 确认 Base64 编码：
    base64.b64encode(f"{user}:{pwd}".encode()).decode()
    base64.b64encode(f"{user}:{pwd}".encode()).decode()
  3. 确认 
    header.Authorization
    header.Authorization
    值以 
    Basic
    Basic
    开头（注意空格）

表不存在（
NoSuchNamespaceException
NoSuchNamespaceException
）

  1. 确认 
    header.Workspace
    header.Workspace
    和 
    default-namespace
    default-namespace
    拼写正确
  2. 先用 
    SHOW TABLES IN cz.public
    SHOW TABLES IN cz.public
    确认表名
  3. 检查账号是否有该 workspace 的访问权限
通过SQL进行数据转换的一些提示
数据分析与SQL指南
Yunqi © 2024 Yunqi, Inc. All Rights Reserved.
联系我们
预约咨询
微信咨询
电话咨询
邮件咨询