当前位置：首页 > news >正文

用 Apache Iceberg 与 Apache Spark 在 Google Cloud 打造高性能、可扩展的数据湖仓

news 2025/8/13 5:52:08

大数据分析版图不断演进，组织正寻求更灵活、可扩展且具成本效益的方式来管理与分析海量数据。这一追求催生了数据湖仓范式，它将数据湖的低成本存储与灵活性，与数据仓库的数据管理能力和事务一致性相结合。站在这场变革中心的是 Apache Iceberg 等开放表格式与 Apache Spark 等强大处理引擎，而这些都运行在坚实可靠的 Google Cloud 基础设施之上。

Apache Iceberg 的崛起：数据湖的变革者

多年来，构建在云对象存储（如 Google Cloud Storage，GCS）上的数据湖具备无与伦比的可扩展性与成本效率。然而，它们常常缺少传统数据仓库中的关键特性，比如事务一致性、模式（Schema）演进，以及面向分析查询的性能优化。Apache Iceberg 正是在这一点上大放异彩。

Apache Iceberg 是一种开放表格式，旨在解决上述局限。它位于云存储中的数据文件（如 Parquet、ORC 或 Avro）之上，通过一层元数据将文件集合转化为高性能、类 SQL 的表。Iceberg 的强大之处在于：

ACID 合规：Iceberg 为数据湖带来原子性、一致性、隔离性和持久性（ACID）属性。这意味着数据写入是事务性的，即便有并发操作也能确保数据完整性，不再出现部分写入或读到不一致数据的问题。
模式演进：在传统数据湖中，管理模式变更是令人头疼的大问题。Iceberg 无缝处理模式演进，允许在不重写底层数据的情况下添加、删除、重命名或重排列，这对敏捷数据开发至关重要。
隐藏分区：Iceberg 智能管理分区，屏蔽数据物理布局。用户无需了解分区方案即可编写高效查询，并且你可以随时间演进分区策略而无需进行数据迁移。
时光回溯与回滚：Iceberg 维护完整的表快照历史。这使“时光回溯”查询成为可能，可以按任意过去时间点查询数据；同时也提供回滚能力，让你将表恢复到此前的良好状态，这对调试与数据恢复价值巨大。
性能优化：Iceberg 丰富的元数据让查询引擎能够高效裁剪无关的数据文件与分区，大幅加速查询执行。它避免代价高昂的文件列举操作，能够基于元数据直接定位相关数据。

借助在数据湖之上提供这些“类数据仓库”能力，Apache Iceberg 使真正的“数据湖仓”成为可能：既拥有云存储的灵活性和成本优势，又具备结构化表的可靠性和性能。

BigQuery 中面向 Apache Iceberg 的 BigLake 表在 Google Cloud 上提供了类似标准 BigQuery 表的全托管体验，但所有数据均存放于客户自有的存储桶中。支持的特性包括：

通过 GoogleSQL 数据操作语言（DML）进行表变更
使用 Storage Write API 统一批处理与高吞吐流式写入（通过 Spark 等 BigLake 连接器）
Iceberg V2 快照导出，并在每次表变更时自动刷新
通过模式演进更新列元数据
自动存储优化
历史数据的时光回溯访问
列级安全与数据脱敏

以下是使用 GoogleSQL 创建一个空的 BigLake Iceberg 表的示例：

SQL

CREATE TABLE PROJECT_ID.DATASET_ID.my_iceberg_table ( name STRING, id INT64 ) WITH CONNECTION PROJECT_ID.REGION.CONNECTION_ID OPTIONS ( file_format = 'PARQUET', table_format = 'ICEBERG', storage_uri = 'gs://BUCKET/PATH' );

随后你可以通过 LOAD DATA 将文件中的数据导入该表，或使用 INSERT INTO 从其他表写入数据。

SQL

从文件加载

LOAD DATA INTO PROJECT_ID.DATASET_ID.my_iceberg_table FROM FILES ( uris = ['gs://bucket/path/to/data'], format = 'PARQUET' );

从表加载

INSERT INTO PROJECT_ID.DATASET_ID.my_iceberg_table SELECT name, id FROM PROJECT_ID.DATASET_ID.source_table

除了全托管形态，Apache Iceberg 还可在 BigQuery 中作为只读外部表使用。可将其指向已有数据文件路径。

SQL

CREATE OR REPLACE EXTERNAL TABLE PROJECT_ID.DATASET_ID.my_external_iceberg_table WITH CONNECTION PROJECT_ID.REGION.CONNECTION_ID OPTIONS ( format = 'ICEBERG', uris = ['gs://BUCKET/PATH/TO/DATA'], require_partition_filter = FALSE );

Apache Spark：数据湖仓分析引擎

Apache Iceberg 为数据湖仓提供结构与管理，而 Apache Spark 则是让其“动起来”的处理引擎。Spark 是一个强大的开源分布式处理系统，以其速度、通用性以及处理多样化大数据工作负载的能力而闻名。Spark 的内存计算能力、丰富的生态（覆盖机器学习与基于 SQL 的处理）以及对 Iceberg 的深度支持，使其成为理想之选。

Apache Spark 深度集成于 Google Cloud 生态。将 Spark 运行在 Google Cloud 上的优势包括：

使用 Google Cloud Serverless for Apache Spark，获得无需集群运维的真正无服务器体验
通过 Dataproc 获得全托管 Spark 体验，并可灵活配置与管理集群
使用 Apache Spark Lightning Engine（预览）加速 Spark 作业
预装 GPU 与驱动的运行时配置
在 Spark 运行时中开箱即用多种 AI/ML 库，包含 XGBoost、PyTorch 与 Transformers
在 BigQuery Studio 中通过 Colab Enterprise 笔记本直接编写 PySpark 代码，并借助 Gemini 生成 PySpark 代码
轻松连接 BigQuery 原生表、BigLake Iceberg 表、外部表与 GCS 中的数据
与 Vertex AI 集成，支持端到端 MLOps

Iceberg + Spark：强强联合

Iceberg 与 Spark 搭配，为构建高性能、可靠的数据湖仓提供强大组合。Spark 可利用 Iceberg 的元数据优化查询计划、高效裁剪数据，并在数据湖范围内确保事务一致性。

你的 Iceberg 表与 BigQuery 原生表可通过 BigLake Metastore（BigLake 元存储）访问。该方式将具备 BigQuery 兼容性的表暴露给开源引擎（包括 Spark）。

Python

from pyspark.sql import SparkSession

创建 Spark 会话

spark = SparkSession.builder
.appName("BigLake Metastore Iceberg")
.config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog")
.config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID")
.config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION")
.config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY")
.getOrCreate() spark.conf.set("viewsEnabled", "true")

使用 blms_catalog

spark.sql("USE CATALOG_NAME;") spark.sql("USE NAMESPACE DATASET_NAME;")

为临时结果配置 Spark

spark.sql("CREATE namespace if not exists MATERIALIZATION_NAMESPACE") spark.conf.set("materializationDataset", "MATERIALIZATION_NAMESPACE")

列出数据集中的表

df = spark.sql("SHOW TABLES;") df.show()

查询表

sql = """SELECT * FROM DATASET_NAME.TABLE_NAME""" df = spark.read.format("bigquery").load(sql) df.show()

sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME""" df = spark.read.format("bigquery").load(sql) df.show()

sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME""" df = spark.read.format("bigquery").load(sql) df.show()

在 BigLake Metastore 的基础上，Iceberg REST Catalog（预览版）可将 Iceberg 数据暴露给任意数据处理引擎。以下是在 Spark 中连接的示例：

Python

import google.auth from google.auth.transport.requests import Request from google.oauth2 import service_account import pyspark from pyspark.context import SparkContext from pyspark.sql import SparkSession

catalog = "" spark = SparkSession.builder.appName("")
.config("spark.sql.defaultCatalog", catalog)
.config(f"spark.sql.catalog.{catalog}", "org.apache.iceberg.spark.SparkCatalog")
.config(f"spark.sql.catalog.{catalog}.type", "rest")
.config(f"spark.sql.catalog.{catalog}.uri", "https://biglake.googleapis.com/iceberg/v1beta/restcatalog")
.config(f"spark.sql.catalog.{catalog}.warehouse", "gs://")
.config(f"spark.sql.catalog.{catalog}.token", "")
.config(f"spark.sql.catalog.{catalog}.oauth2-server-uri", "https://oauth2.googleapis.com/token")
.config(f"spark.sql.catalog.{catalog}.header.x-goog-user-project", "")
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config(f"spark.sql.catalog.{catalog}.io-impl", "org.apache.iceberg.hadoop.HadoopFileIO")
.config(f"spark.sql.catalog.{catalog}.rest-metrics-reporting-enabled", "false")
.getOrCreate()

完善湖仓版图

Google Cloud 提供一套与 Apache Iceberg 与 Apache Spark 相辅相成的完整服务，帮助你在沿用开源技术栈的同时，轻松构建、管理与扩展数据湖仓：

Dataplex 通用目录（Universal Catalog）：提供跨数据湖、数据仓库与数据集市的统一数据织体，用于管理、监控与治理数据。它与 BigLake Metastore 集成，确保治理策略在 Iceberg 表上得到一致执行，并支持语义检索、数据血缘与数据质量检查等能力。
Google Cloud 托管版 Apache Kafka：在 Google Cloud 上运行全托管的 Kafka 集群（含 Kafka Connect）。数据流可直接写入 BigQuery，包括以低延迟读取托管的 Iceberg 表。
Cloud Composer：基于 Apache Airflow 的全托管工作流编排服务。
Vertex AI：用于管理端到端的 MLOps 体验。你也可以使用 Vertex AI Workbench（托管 JupyterLab）连接到无服务器 Spark 与 Dataproc 实例。