01

启动 Notebook

首先，如果您还没有 Docker 和 Docker Compose ，请先安装它们。接下来，创建一个包含以下内容的 docker-compose.yaml 文件。

docker-compose.yaml

version: "3"

services:
  spark-iceberg:
    image: tabulario/spark-iceberg
    depends_on:
      - postgres
    container_name: spark-iceberg
    environment:
      - SPARK_HOME=/opt/spark
      - PYSPARK_PYTON=/usr/bin/python3.9
      - PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/opt/spark/bin
    volumes:
      - ./warehouse:/home/iceberg/warehouse
      - ./notebooks:/home/iceberg/notebooks/notebooks
    ports:
      - 8888:8888
      - 8080:8080
      - 18080:18080
  postgres:
    image: postgres:13.4-bullseye
    container_name: postgres
    environment:
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=demo_catalog
    volumes:
      - ./postgres/data:/var/lib/postgresql/data

在 docker-compose.yaml 同级目录中，运行以下命令以启动并打开一个支持 Iceberg 的 Spark notebook 服务。

docker-compose up -d
docker exec -it spark-iceberg pyspark-notebook

就是这么简单！

您现在应该有了：

• 一个运行在 http://localhost:8888 的功能齐全的 notebook 服务

• 一个运行在 http://localhost:8080 的 Spark driver

• 一个运行在 http://localhost:18080 的 Spark history

  
  

一个最小化的运行环境

docker compose 文件提供的运行环境远不是一个大规模的生产级仓库，但它确实可以让你演示 Iceberg 的诸多特性。让我们快速介绍一下这个最小化的运行环境：

02

那么，Iceberg能做什么呢？

本文的 docker 环境附带一个名为 “ Iceberg - Getting Started ” 的 notebook，其演示了本文将会深入介绍的大量特性。您在此处看到的示例同样包含在 notebook 中，因此它也是开始学习的好地方！

模式演变

任何使用过 Hive 表或其他大数据表格式的人都知道改变表模式会是多么棘手。添加先前已删除的列可能会导致旧的列数据死而复生。即便您幸运地快速捕获了它，撤消操作通常会非常耗时，并且常常会导致您回到表的原始状态。而即使只修改列名也是一个潜在的危险操作。

在 Iceberg 中，您不必担心哪些更改生效，哪些更改会破坏您的表。在 Iceberg 中，重命名或添加列等表模式操作是安全的，没有令人始料未及的副作用。

ALTER TABLE taxis
RENAME COLUMN fare_amount TO fare

ALTER TABLE taxis
ADD COLUMN fare_per_distance_unit float AFTER distance;

其他模式更改（例如更改列的类型、添加注释或移动其位置）也同样易如反掌。

ALTER TABLE taxis RENAME COLUMN trip_distance TO distance;
ALTER TABLE taxis ALTER COLUMN distance COMMENT 'The elapsed trip distance in miles reported by the taximeter.'
ALTER TABLE taxis ALTER COLUMN distance TYPE double;
ALTER TABLE taxis ALTER COLUMN distance AFTER fare;

此外，因为模式演变仅包括元数据操作，所以它们执行得非常迅速。仅仅为了更改单个列的名称而重写整个表的日子已经一去不复返了！

隐藏分区

当您考虑更改生产表的分区方式时，这总是一件大事。这个过程很艰巨，需要迁移数据到具有不同分区方式的全新表——除非您使用的是 Iceberg。历史数据必须变更为新的分区方式，即使您只想对新的数据变更分区方式。可以说，最困难的是追查表的使用者，以提醒他们在所有查询中更新分区子句！Iceberg 可以无缝地处理这个问题。表的分区可以实时更新并仅应用于新写入的数据。

ALTER TABLE taxis
ADD PARTITION FIELD VendorID

而查询计划（ query plans ）将会被拆分开来，分区变更之前写入的数据使用旧的分区方式，变更之后写入的数据使用新的分区方式。查询表的人甚至无需感知拆分动作。WHERE 子句中的简单谓词会自动转换为分区过滤器，以裁剪不匹配的文件。这就是 Iceberg 中所说的 隐藏分区。

时间旅行和回滚

当您更改表时，Iceberg 会及时地将每个版本作为“快照”进行跟踪。您可以时间旅行到任何快照或时间点。当您想要重现先前查询的结果时，这一特性会非常有用，比如重新生成一个下游报表产品。

spark.read.table("taxis").count()  # 2,853,020

val ONE_DAY_MS = 86400000;
val NOW = System.currentTimeMillis()

(spark
.read
.option("as-of-timestamp", NOW_MS - ONE_DAY_MS)
.table("taxis")
.count())  # 2,798,371

在一个依赖于许多会频繁更新的上游表的复杂逻辑中，引起查询结果变动的原因，是正在测试的新代码还是上游表的更改，往往不好精准确认。而使用时间旅行，您可以确保查询到指定时间下的所有表（译者注：指定时间下的表数据不会因新的更改而变动），这使创建受控的环境变得更加容易。

如果您不想在运行时对单个查询进行时间旅行，而是希望将表实际回滚到特定时间点或特定快照 ID，您可以使用回滚过程语句来轻松实现！

CALL catalog_name.system.rollback_to_timestamp('taxis', TIMESTAMP '2021-12-31 00:00:00.000')

CALL demo.system.rollback_to_snapshot('taxis', <SNAPSHOT>)

极富表现力的行级SQL

Iceberg 的扩展 SQL 可以执行应用于行级操作的查询，它非常有表现力。例如，您可以删除表中与特定谓词匹配的所有记录。

DELETE FROM taxis
WHERE fare_per_distance_unit > 4.0 OR distance > 2.0

此外，MERGE INTO 语句使合并两个表的任务非常直观。

MERGE INTO prod.nyctaxis pt
USING (SELECT * FROM staging.nyc.taxis) st
ON pt.id = st.id
WHEN NOT MATCHED THEN INSERT *;

原子性的 CTAS 和 RTAS 语句

[CREATE|REPLACE] TABLE prod.nyc.vendor2 AS
SELECT * FROM taxis
WHERE vendor_id = '2'

Spark 过程语句

结束语

End