Snowflake托管Iceberg表的事件驱动Glue Catalog注册方案

在没有 Glue 爬虫的情况下，以近乎实时的方式消除随机后缀问题，每天只需几美分。

我们的每日站会开始了，客户直接开门见山地说：“我把你昨天创建的 Iceberg 表添加到了 Glue Catalog，我想知道你能不能在建表时不要在表名末尾加上那个随机后缀。”

客户解释说，结果并不是简洁的名称，而是像 customers_f0xgvx1a 和 orders_djctgenc 这样的名字。对于最初编写 SQL 来探索数据的用户来说，这些名称既繁琐又令人沮丧。

Snowflake 在 BCR 捆绑版本 2025_01 中为 Snowflake 托管的 Iceberg 表引入了新的写入路径。此功能无法禁用，它会在基础位置后附加一个随机的 8 字符后缀，以便 Snowflake 在表被删除并重新创建时保证唯一性。Glue 爬虫只是从元数据中读取这个名字并将其写入到目录中。

我心里想：“没问题。”Glue 中肯定有一个设置可以关掉这个功能，或者有一条规则可以用来重命名表以去掉后缀。

不幸的是，Glue 中并不存在这些功能。由于没有获得采用新目录服务的批准，也不想重新架构那些对其他所有数据集都已经行之有效的访问模式，我们不得不另辟蹊径。在经历了几次失败的尝试和走入死胡同的变通方案之后，我最终决定直接将表注册到 Glue 目录中。

我们的管道运行良好。由 Snowflake 管理我们的 Iceberg 表和元数据，我们已经完成了成功交付的 95%。我们只需要拿下最后那 5%，而事件驱动的注册就是答案。

在构建此方案之前：请考虑 Horizon Catalog

企业通常受限于他们可以使用的工具和 API，不能仅仅因为想用新工具就用，还需要获得组织内各方人员的批准。如果那些选项不可行，那么这篇文章将对你有所帮助。

本文所述的自定义解决方案适用于 1) 尚未采用 Horizon Catalog 或其他符合 REST API 规范的目录的组织。2) 你在受监管的环境中工作，添加新的目录服务需要漫长的审批流程。3) 你的下游消费者严格锁定使用 AWS Glue 以获取 Lake Formation 权限，或者已有 Athena 工作负载，和/或 4) 你需要在今天就得到可用的东西，而不能等待高层平台决策。

解决方案：无需爬虫的事件驱动注册

扫描表元数据的爬虫可能是一种自然的替代方案，但此解决方案相比使用爬虫有几个好处。其中之一是我们在 Snowflake 将元数据文件写入 S3 的那一刻就直接将表注册到 Glue Data Catalog 中。SQS 位于 S3 和 Lambda 之间，因为它将事件源与处理解耦，支持针对调用失败的死信队列，并避免在多个表同时写入时触及 Lambda 对 S3 的直接并发限制。该架构的关键组件如下：

驱动此架构的关键原则：

• 无爬虫。我们使用直接 API 注册，近乎实时。
• 简洁名称。我们严格控制注册到 Glue Catalog 中的表名。
• 完整模式提取。我们直接从 Iceberg metadata.json 中提取列，而不是硬编码它们。
• 自动模式演进。新列会在下一次写入时出现在 Glue 中。
• 幂等性。同一张表的多个事件只会更新指针。

实现

Lambda 函数处理核心逻辑。它通过 SQS 接收 S3 事件，解析路径以获取简洁的表名，从 Iceberg 元数据中读取模式，并将其注册到 Glue 中。

路径解析： 随机后缀正好是 8 个字符的混合大小写字母数字。

PATH_PATTERN = re.compile(
    r'^(?P<prefix>.+)/(?P<base_loc>[^/]+)\.(?P<rand>[A-Za-z0-9]{8})'
    r'/metadata/(?P<file>.+\.metadata\.json)$')

对于像 iceberg/lseg_qa_iceberg/CUSTOMERS.f0XGVx1a/metadata/00042-uuid.metadata.json 这样的路径，基础位置变成了 CUSTOMERS。我们保留这个简洁名称并丢弃随机的 f0XGVx1a 字符串。

def extract_columns_from_metadata(bucket, metadata_key):
    obj = s3.get_object(Bucket=bucket, Key=metadata_key)
    metadata = json.loads(obj['Body'].read())
    # Iceberg v2: find current schema by ID
    current_schema_id = metadata.get('current-schema-id', 0)
    schemas = metadata.get('schemas', [])
    schema = next(
        (s for s in schemas if s.get('schema-id') == current_schema_id),
        schemas[-1] if schemas else None
    )
    columns = []
    for field in schema.get('fields', []):
        columns.append({
            'Name': field['name'].lower(),
            'Type': iceberg_type_to_hive(field['type']),
        })
    return columns

Iceberg metadata.json 包含完整的表模式，我们直接从 S3 读取。我们解析该文件，然后通过 iceberg_type_to_hive 将数据类型映射到 Hive 等效类型。请注意，此函数必须处理基本类型（string、long、double、boolean、date、timestamptz）以及复杂类型——结构体、列表和映射——这些需要递归转换。Decimal 类型必须保留精度和小数位数（例如 decimal(18,2)）。Athena v3、Trino 和现代 Spark 都接受这些 Hive 风格的类型字符串。iceberg_type_to_hive 的参考实现可在本文末尾链接的源代码中找到。由于这些文件只有几 KB，我们完全避免了扫描海量数据文件或手动维护任何列列表。

Glue 注册（带重试）：

def register_table(table_name, table_location, metadata_location, columns):
    table_input = {
        'Name': table_name,
        'TableType': 'EXTERNAL_TABLE',
        'Parameters': {
            'table_type': 'ICEBERG',
            'metadata_location': metadata_location,
            'classification': 'iceberg',
        },
        'StorageDescriptor': {
            'Columns': columns,
            'Location': table_location,
            'InputFormat': 'org.apache.iceberg.mr.hive.HiveIcebergInputFormat',
            'OutputFormat': 'org.apache.iceberg.mr.hive.HiveIcebergOutputFormat',
            'SerdeInfo': {
                'SerializationLibrary': 'org.apache.iceberg.mr.hive.HiveIcebergSerDe'
            },
        }
    }
    max_retries = 3
    for attempt in range(max_retries):
        try:
            try:
                glue.get_table(DatabaseName=GLUE_DATABASE, Name=table_name)
                glue.update_table(DatabaseName=GLUE_DATABASE, TableInput=table_input)
            except glue.exceptions.EntityNotFoundException:
                glue.create_table(DatabaseName=GLUE_DATABASE, TableInput=table_input)
            return
        except ClientError as e:
            if e.response['Error']['Code'] == 'ConcurrentModificationException' \
            and attempt < max_retries - 1:
                time.sleep((attempt + 1) * 0.5)
            else:
                raise

传入 metadata_location 参数使这一点得以实现。Athena v3、Spark 和 Trino 等查询引擎会从 Glue 表条目中读取此字段，并直接转到 Iceberg metadata.json 以获取模式和快照数据。我们填充 Columns 列表严格来说是为了向后兼容旧工具，如 Hive 3.x、较旧的 Presto 配置，或任何从 Glue 的 StorageDescriptor 而非直接从 Iceberg 元数据读取模式的工具。Athena v3、带 Iceberg 连接器的 Spark 和 Trino 都会忽略 StorageDescriptor 列，而选用原生 Iceberg 模式。 注意：此示例使用旧版 Hive 格式类（HiveIcebergInputFormat / HiveIcebergOutputFormat / HiveIcebergSerDe）以获得与旧工具的最大兼容性。如果你不需要 Hive 3.x 或旧版 Presto 兼容性，AWS Glue 现在支持通过 OpenTableFormatInput / IcebergInput API 原生注册 Iceberg 表，这完全跳过了 Hive SerDe，是未来更现代的方法。

S3 事件配置你必须按 .metadata.json 后缀进行过滤。这是一个关键细节。Snowflake 在每次刷新时都会写出 Parquet 数据文件、Avro 清单文件和 metadata.json 文件。只有 metadata.json 才代表一次已完成的提交。如果不加过滤，你的 Lambda 会在每次文件写入时触发，并用不匹配路径的警告淹没你的日志。

事件名称：iceberg-metadata-notify 前缀：iceberg/my_schema/ 后缀：.metadata.json 事件类型：s3:ObjectCreated:* 目标：SQS 队列

IAM 策略为 Lambda 角色提供最小权限：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": ["glue:GetTable", "glue:CreateTable", "glue:UpdateTable"],
            "Resource": [
                "arn:aws:glue:REGION:ACCOUNT:catalog",
                "arn:aws:glue:REGION:ACCOUNT:database/my_db",
                "arn:aws:glue:REGION:ACCOUNT:table/my_db/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject"],
            "Resource": ["arn:aws:s3:::my-bucket/*"]
        },
        {
            "Effect": "Allow",
            "Action": ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes"],
            "Resource": ["arn:aws:sqs:REGION:ACCOUNT:my-queue"]
        }
    ]
}

已验证的内容

这当然不是一份详尽的列表，但以下是我为确保直接注册成功而运行的验证及其预期结果。

成本与性能

运行此 Lambda 的财务成本可以忽略不计。你为每张表每次 DML 提交支付一次调用费用，而每次调用只读取几 KB 的 JSON。一个每 5 分钟刷新 20 张表的示例管道每天大约会触发 5,760 次调用，而 Lambda、SQS 和 S3 GetObject 费用合计每天大约只需 $0.01–$0.02。

真正的价值来自你无需再付费的部分。

• 无需 Glue 爬虫计算资源。
• 无需爬虫调度/管理开销。
• 爬虫运行之间不存在数据陈旧窗口。
• 近乎实时的可用性，端到端通常不到一分钟（S3 事件通知投递到 SQS 加上 Lambda 轮询大约增加 30–60 秒），而不是几分钟。

何时使用哪种方法

如果你的组织允许使用 Horizon Catalog，那就是应该走的路。Horizon Catalog 无疑是平台原生的答案，但如果你目前仍困在以 Glue 为中心的环境中，这种事件驱动的 Lambda 模式非常有效。它将提供干净的命名、自动模式同步和近乎实时的可用性，且运维开销很小。

结论

通过使用原生 S3 事件通知来处理你的外部目录更新，并将元数据提交视为从事实来源读取模式的触发器，你可以维持下游工具所需的精确注册。

这个基于 Python 的 Lambda 解决方案部署快速，每月运营成本很低。主要好处在于用户体验：分析师可以执行像 SELECT * FROM customers 这样的简单查询，而无需处理 Snowflake 管理的 Iceberg 表所生成的后缀相关不一致问题。

我们在保留 Snowflake 管理的 Iceberg 优势的同时，也为前面提到的解决方案（如 Horizon Catalog）搭建了一座桥梁，使其能够可用。

详细的设置指南、CloudFormation 模板和完整源代码可按需提供。

针对 Snowflake 管理的 Iceberg 表的事件驱动 Glue Catalog 注册最初发布于 Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science 在 Medium 上，人们正在那里通过高亮和回应这个故事来继续对话。