从 Basin Catalog 或任何 Apache Iceberg catalog 删除数据时,必须通过 catalog 本身以事务方式执行操作。直接手动删除元数据或数据文件可能导致数据 catalog 损坏。
Basin Catalog 可以自动管理快照过期和压缩等表维护操作。这些持续运行的操作有助于降低延迟和存储成本。
- 快照过期:自动删除旧快照及相应的未引用数据文件。这同时减少了元数据开销和存储成本。
- 压缩:将小型数据文件合并为更大的文件。这优化了读取性能,并减少了查询期间读取的文件数量。
如果不启用自动维护,则需要手动处理这些操作。
在表维护文档中了解更多信息。
为 catalog 启用自动快照过期:
npx wrangler basin catalog snapshot-expiration enable my-bucket --older-than-days 30 --retain-last 5yarn wrangler basin catalog snapshot-expiration enable my-bucket --older-than-days 30 --retain-last 5pnpm wrangler basin catalog snapshot-expiration enable my-bucket --older-than-days 30 --retain-last 5为 catalog 启用自动压缩:
npx wrangler basin catalog compaction enable my-bucket --target-size 256yarn wrangler basin catalog compaction enable my-bucket --target-size 256pnpm wrangler basin catalog compaction enable my-bucket --target-size 256请参阅管理 catalog文档中的更多示例。
在以下情况下需要手动删除数据:
- 遵守 GDPR 或 CCPA 等数据保留政策。
- 使用条件逻辑进行选择性删除。
- 移除 Basin Catalog 未管理的陈旧或未引用文件。
以下示例使用 PySpark,但也可以使用其他兼容 Iceberg 的引擎执行类似操作。要配置 PySpark,请参阅我们的示例或官方 PySpark 文档 ↗。
# Creates new snapshots and marks old files for cleanup
spark.sql("""
DELETE FROM r2dc.namespace.table_name
WHERE column_name = 'value'
""")
# The following is effectively a TRUNCATE operation
spark.sql("DELETE FROM r2dc.namespace.table_name")
# For large deletes, use partitioned tables and delete entire partitions for faster performance:
spark.sql("""
DELETE FROM r2dc.namespace.table_name
WHERE date_partition < '2024-01-01'
""")# Removes table from catalog but keeps data files in R2 storage
spark.sql("DROP TABLE r2dc.namespace.table_name")
# ⚠️ DANGER: Permanently deletes all data files from R2
# This operation cannot be undone
spark.sql("DROP TABLE r2dc.namespace.table_name PURGE")
# Use CASCADE to drop all tables within the namespace
spark.sql("DROP NAMESPACE r2dc.namespace_name CASCADE")
# You will need to PURGE the tables before running CASCADE to permanently delete data files
# This can be done with a loop over all tables in the namespace
tables = spark.sql("SHOW TABLES IN r2dc.namespace_name").collect()
for row in tables:
table_name = row['tableName']
spark.sql(f"DROP TABLE r2dc.namespace_name.{table_name} PURGE")
spark.sql("DROP NAMESPACE r2dc.namespace_name CASCADE")# Remove old metadata and data files marked for deletion
# The following retains the last 5 snapshots and deletes files older than Nov 28, 2024
spark.sql("""
CALL r2dc.system.expire_snapshots(
table => 'r2dc.namespace_name.table_name',
older_than => TIMESTAMP '2024-11-28 00:00:00',
retain_last => 5
)
""")
# Removes unreferenced data files from R2 storage (orphan files)
spark.sql("""
CALL r2dc.system.remove_orphan_files(
table => 'namespace.table_name'
)
""")
# Rewrite data files with a target file size (e.g., 512 MB)
spark.sql("""
CALL r2dc.system.rewrite_data_files(
table => 'r2dc.namespace_name.table_name',
options => map('target-file-size-bytes', '536870912')
)
""")Apache Iceberg 使用分层元数据结构来高效管理表数据。以下是关键组件和文件结构:
- metadata.json:指向当前快照的顶层 JSON 文件
- snapshot-*:给定时间点的不可变表状态
- manifest-list-*.avro:列出给定快照所有 manifest 文件的 Avro 文件
- manifest-file-*.avro:跟踪数据文件及其统计信息的 Avro 文件
- data-*.parquet:包含实际表数据的 Parquet 文件
- 注意:未更改的 manifest 文件会在快照之间复用
- metadata.json 元数据文件 - 指向当前快照
- 表 schema
- 分区规范
- 排序顺序
- 快照
- snapshot-3051729675574597004.avro 快照 1(历史)
- manifest-list-abc123.avro manifest 列表
- manifest-file-001.avro manifest 文件
- data-00001.parquet(10 MB,5 万行)
- data-00002.parquet(12 MB,6 万行)
- data-00003.parquet(11 MB,5.5 万行)
- manifest-file-002.avro
- data-00004.parquet(9 MB,4.5 万行)
- data-00005.parquet(10 MB,5 万行)
- manifest-file-001.avro manifest 文件
- manifest-list-abc123.avro manifest 列表
- snapshot-3051729675574597005.avro 快照 2(当前)
- manifest-list-def456.avro manifest 列表
- manifest-file-001.avro (从快照 1 复用)
- data-00001.parquet
- data-00002.parquet
- data-00003.parquet
- manifest-file-003.avro (新)
- data-00006.parquet(11 MB,5.3 万行)
- data-00007.parquet(10 MB,5.1 万行)
- data-00008.parquet(12 MB,5.8 万行)
- manifest-file-001.avro (从快照 1 复用)
- manifest-list-def456.avro manifest 列表
- snapshot-3051729675574597004.avro 快照 1(历史)
Apache Iceberg 支持两种删除模式:Copy-on-Write (COW) 和 Merge-on-Read (MOR)。两者都会创建新快照并将旧文件标记为待清理,但处理删除的方式不同:
| 方面 | Copy-on-Write (COW) | Merge-on-Read (MOR) |
|---|---|---|
| 删除方式 | 重写不含已删除行的数据文件 | 创建 delete 文件标记要跳过的行 |
| 查询性能 | 快(无需合并) | 较慢(需要读取时合并) |
| 写入性能 | 较慢(重写数据文件) | 快(仅写入 delete 标记) |
| 存储影响 | 立即创建新数据文件 | 随时间累积 delete 文件 |
| 维护需求 | 快照过期 | 快照过期 + 压缩(rewrite_data_files) |
| 适用场景 | 读密集型工作负载 | 频繁小变更的写密集型工作负载 |
这些操作对 COW 和 MOR 表的工作方式相同:
| 操作 | 作用 | 数据已删除? | 可撤销? |
|---|---|---|---|
DELETE FROM |
删除符合条件的行 | 否(标记为待清理) | 通过时间旅行1 |
DROP TABLE |
从 catalog 中移除表 | 否 | 是(如果数据文件仍存在) |
DROP TABLE ... PURGE |
移除表并删除数据 | 是 | 否 |
expire_snapshots |
清理旧快照/文件 | 是 | 否 |
remove_orphan_files |
移除未引用的文件 | 是 | 否 |
对于 Merge-on-Read 表,可能需要手动应用删除以提升性能:
| 操作 | 作用 | 使用时机 |
|---|---|---|
rewrite_data_files (compaction) |
应用删除并合并文件 | 当大量 delete 文件导致查询性能下降时 |
- 表维护 - 了解自动维护操作
- Basin Catalog - 概览和快速入门指南
- 查询数据 - 使用 Basin SQL 查询表
- Apache Iceberg Maintenance ↗ - Iceberg 官方表维护文档
-
时间旅行功能在调用
expire_snapshots之前可用 ↩