用Apache Iceberg和Flink构建动态流式数据湖

处理上游模式更改是流式数据管道写入数据湖时常见的运维挑战。当源模式发生变化时，团队常常面临两难选择：重启管道或执行手动迁移。重启可能会暂停数据摄取，并延迟或丢失在途数据。手动迁移会消耗工程时间，并在数据湖落后于源时引入模式不一致的风险。

例如，考虑一个Apache Flink作业，它摄取order_events并写入Iceberg表。周一，管道正常运行。到周三，上游团队添加了一个新的loyalty_tier字段，并引入了一种新的interaction_events事件类型。传统上，您需要停止Flink作业，更新模式定义，然后重新部署。使用Apache Iceberg 动态Iceberg接收器在Amazon Managed Service for Apache Flink上，管道可以在记录级别处理这两种更改而不会中断。DynamicSink将每个事件路由到正确的Iceberg表，并在新列出现时演进表模式，无需操作员干预。

Managed Service for Apache Flink是一项完全托管的AWS服务，您可以使用它来构建和部署流应用程序，而无需设置基础设施和管理资源。Apache Flink的分布式处理引擎通过检查点提供精确一次处理保证，与Apache Iceberg的两阶段提交配合，提供端到端一致性，无重复或无数据丢失。

在本文中，我们向您展示如何构建一个动态流式数据湖，该数据湖能适应新事件类型和模式更改，而无需停止管道。使用Managed Service for Apache Flink上的Apache Flink 2.3和Apache Iceberg 1.11.0，我们逐步介绍了用于逐记录表路由和自动模式演进的DataStream API模式。完整实现在此GitHub仓库中提供。

Apache Iceberg动态接收器

“动态Iceberg接收器允许Flink根据用户定义的逻辑，动态地将记录路由到多个Iceberg表。它还可以在流执行期间动态创建和更新表，并演进表模式和分区规范，通过DynamicRecord类控制，这消除了需求变化时重启Flink作业的需要。

使用DynamicIcebergSink进行逐记录表路由

DynamicIcebergSink在记录级别而不是在管道配置时解析目标表。记录流经DynamicRecordGenerator，对于每个输入，它会发出一个或多个DynamicRecord值。每个DynamicRecord携带其自己的目标表ID、模式、分区规范和行负载，因此接收器知道在哪里写入以及表的外观：

DynamicIcebergSink.forInput(events)
    .generator(generator)
    .catalogLoader(catalogLoader)
    .immediateTableUpdate(true)
    .cacheMaxSize(cacheMaxSize)
    .cacheRefreshMs(cacheRefreshMs)
    .append();

生成器接收每条记录，并发出一个针对已解析表的DynamicRecord，该表如下所示：

return new DynamicRecord(
    tableId,
    tableBranch,
    icebergSchema,
    rowData,
    partitionSpec,
    distributionMode,
    1);

如果表不存在，接收器会创建它，并在记录携带新列时演进其模式。cacheMaxSize和cacheRefreshMs限制了接收器按表的元数据缓存，因此写入多个表的作业不会在每条记录上重新加载元数据。immediateTableUpdate(true)控制这些目录更改如何应用，以下关于自动模式演进的部分将对此进行说明。单个Flink作业可以摄取和路由order_events、interaction_events、user_events以及未来的事件类型，而无需额外的接收器定义。

但是，接收器还需要知道表的外观。这就是为什么每个DynamicRecord还携带Iceberg模式，以便DynamicIcebergSink可以在首次看到时创建表，并在新字段出现时演进它。模式信息可以从数据中推断或从模式注册表中读取。

自动模式演进

流式源随时间添加新字段，DynamicIcebergSink无需重启即可处理它们。在写入每条记录之前，它会将记录的模式与目标表进行比较。如果记录有新字段，Iceberg会将其添加为可选列，并在下一次数据文件提交时提交更改。现有文件仍然有效，无需重写表。当您查询较旧文件时，新列返回null。

immediateTableUpdate设置控制目录更改发生的位置。GitHub示例仓库设置immediateTableUpdate=true，因此看到新模式的写入子任务在发出记录之前就内联应用创建或修改。这提供了最低延迟，但对目录进行更多并发调用。当设置为false时，需要表更改的记录会走弯路。表、模式和分区规范已经与接收器缓存元数据匹配的记录直接发送到写入器。确实需要更改的记录会被按键（表名）路由到更新操作符，因此同一表的更新一次仅应用一个。一旦更新提交并且缓存刷新，后续记录再次匹配并跳过弯路。在稳定状态下，没有模式更改到达时，此路径不会增加额外的混洗。无论哪种方式，模式比较和由此产生的表更改都是相同的。

默认情况下，模式更改是非破坏性的。接收器可以添加新列、加宽现有类型（例如，int到long或float到double）、将必需列放宽为可选列，以及删除列。重要的是，DynamicIcebergSink在撰写本文时不支持重命名列。

源模式通过两种方式识别：从源记录推断模式（例如 JSON 推断）和从模式注册表读取序列化记录（例如 AWS Glue 模式注册表 (GSR)）。Iceberg 目标表的模式演变行为取决于模式来源。JSON 推断会添加它看到的任何新字段，没有契约。例如，这允许作业最初将模式推断为整数，然后在检测到更大值时扩展为长整数。模式注册表序列化记录使用注册表的兼容性规则（例如 BACKWARD）定义策略。这意味着不兼容的生产者更改在注册模式时被拒绝，而不是在写入时。

分区规范在每个 DynamicRecord 上传递，因此接收器在创建或更新表时应用它。我们的示例如何派生该规范在分区部分介绍。

解决方案概述

下图说明了解决方案架构。数据生成器（本地 Java 应用程序）将事件写入 Amazon Kinesis Data Streams。在 Avro 模式下，它还在 AWS Glue Schema Registry 中注册每个事件架构。Apache Flink 的托管服务应用程序消费流，为每条记录解析目标 Iceberg 表，并将数据写入 Amazon S3 中的 Iceberg 表，这些表要么在 AWS Glue Data Catalog 中编目，要么对于完全托管的表，在 Amazon S3 Tables（Amazon S3 的一项功能）中编目。

图 1：在 Apache Flink 托管服务上将流记录路由到每事件 Iceberg 表的解决方案架构

在高层次上，一个 Apache Flink 托管服务应用程序从 Kinesis 读取原始记录，并为每条记录解析目标 Iceberg 表。它使用 DynamicIcebergSink 按需创建和发展表。同一个作业处理多种事件类型，因为目标是根据每条记录决定的，而不是根据接收器。

关于流拓扑的说明：示例假设一个 Kinesis 流承载多种事件类型，这使演示保持专注。这不是该模式的必需条件。如果您的事件到达不同的流（例如，每个生产者或每个域一个流），则每个流创建一个 KinesisStreamsSource，并在接收器之前将它们合并到一个 DataStream 中。路由生成器从记录本身选择目标表，因此多个源可以汇入一个 DynamicIcebergSink，仍然落在正确的表中。

合并不会增加洗牌成本。接收器总是通过内部每表写入键重新分配记录，因此合并流和 N 个独立管道在每条记录交换上产生相同的成本。每个 DynamicRecord 携带的分发模式只改变行落在哪个写入子任务上，而不是是否发生洗牌。真正的权衡是隔离。所有表共享一个写入器池、一个提交聚合器和一个提交器。因此，热流的背压和检查点对齐与其他每个流耦合，并且写入器并行度是整个作业范围的设置。当您有许多小型到中型事件类型需要集中容量时，首选一个合并管道。当某个流的数据量足够大，需要自己的写入器并行度和故障隔离时，则拆分为单独的应用程序。

DynamicIcebergSink 需要每条记录的模式。示例提供了两种可互换的获取方式，实现为两种生成器变体：选项 1 在运行时从每条 JSON 记录推断模式。选项 2 从 AWS Glue Schema Registry 读取注册的模式。下游的所有内容（路由、表创建和模式演变）都是相同的，只有生成器不同。

选项 1：从 JSON 记录推断模式

SchemaAgnosticRoutingGenerator 实现 Iceberg 的 DynamicRecordGenerator。其 generate 方法将路由字段映射到表名称，推断模式，派生分区规范，并通过收集器发出 DynamicRecord：

@Override
public void generate(JsonNode json, Collector<DynamicRecord> out) {
    String tableName = determineTableName(json); // routing field -> table name
    TableIdentifier tableId = TableIdentifier.of(database, tableName);
    Schema schema = inferSchemaFromJson(json); // cached by schema signature
    RowData rowData = convertJsonToRowData(json, schema);
    PartitionSpec spec = buildPartitionSpec(schema); // cached per schema
    out.collect(new DynamicRecord(
        tableId, "main", schema, rowData, spec, DistributionMode.NONE, 4));
}

表名称来自显式的 table-name 字段（如果存在），否则来自路由字段（默认为 event_type）。

对于无模式或半结构化 JSON，生成器直接从每条记录推断 Iceberg 模式。这很方便，但推断本质上有损，因为 JSON 不携带类型信息。因此，生成器应用故意保守的规则，并选择稳定的类型而不是最窄的类型：

对路由表进行分区

分区由我们的生成器决定，而不是由接收器决定，并且相同的机制适用于两种模式选项：JSON 推断和模式注册表生成器共享分区候选逻辑。开源 DynamicIcebergSink 应用每个 DynamicRecord 携带的任何 PartitionSpec。我们示例的 SchemaAgnosticRoutingGenerator 在运行时构建该规范：它从分区候选应用程序属性读取候选分区字段列表，并根据它观察到的字段为每个表派生规范。对于每个表，buildPartitionSpec 遍历该列表，并仅保留表模式中存在的候选。

相同的列表适应每个表。具有 event_date 和 region 的表按 identity(event_date) 和 identity(region) 分区。没有任何候选的表创建为未分区。生成的规范在每个 DynamicRecord 上传递，因此接收器在首次创建表时应用它。

例如，使用 partition.candidates = event_time,region,product：一个架构中包含 event_time 和 product 的表将按这两个字段进行分区。仅包含 event_time 的表将使用 identity(event_time)。不包含任何候选字段的表将不分区。分区规格在创建时也不是固定的：sink 通过 Iceberg 分区规格演化来演进它们，当候选字段稍后出现在表架构中时添加，消失时移除。这是仅元数据更改，因此现有数据文件保留其写入时的规格。

接下来是两个操作实践。首先，始终将事件时间字段包含在候选项中，以确保每个表至少按时间分区，并通过表的 $partitions 元数据或目录中的规格监控未分区的表：如果生产者发出 create_timestamp 而不是 event_time，将静默创建未分区表，直到更新候选列表。其次，谨慎处理像 region 这样的通用字段。如果源为候选字段产生高基数数值，您可以稍后更正规格。演化仅适用于新写入的文件，因此已写入的小文件保留到压缩重写。

请注意，候选列表是全局的，不是每表独有的。它跟踪您可能分区的每个字段，而每个表仅使用其拥有的字段。

选项 2：从模式注册表读取架构

推断很方便但有损，且不提供契约：没有什么能阻止生产者静默更改字段的类型或含义。第二个选项通过从注册表而不是数据中读取架构来消除猜测。许多生产流平台标准化了通过 AWS Glue Schema Registry 管理的强类型 Avro 架构。使用 GSR，生产者显式注册架构，Kinesis 上的每条记录均以 Avro 编码并带有架构版本 ID 前缀，消费者根据精确注册的架构解码。这为您提供了 JSON 推断无法提供的三点：精确的类型（long 保持为 long，timestamp-micros 保持为 timestamp-micros）、在注册时强制执行的受治理的演化策略，以及生产者和消费者共享的单一事实来源。

该模式适用于任何为消费者提供每条记录的写入者架构的模式注册表。示例使用 AWS Glue Schema Registry 实现，但相同的生成器形状适用于其他注册表。

dynamic-sink-avro-sample 模块将 GSR 管理的 Avro 架构应用于相同的动态路由和架构演化模式。对于每条记录，AvroToDynamicRecordGenerator 读取架构版本 ID，并从 GSR 获取写入者架构，在首次查找后缓存它。然后将其转换为 Iceberg 架构，将负载解码为 RowData，并发出 DynamicRecord，与 JSON 生成器完全相同：

sink 接线与选项 1 相同。只有生成器变化，由于源携带原始 Avro 字节，输入流是 byte[] 而不是解析后的 JSON：

AvroToDynamicRecordGenerator generator = new AvroToDynamicRecordGenerator(
    awsRegion, registryName, database, partitionCandidates, branch);
DynamicIcebergSink.forInput(eventBytes)
    .generator(generator)
    // identical catalogLoader, immediateTableUpdate(true), cache, and write settings as option 1
    .append();

由于架构来自 GSR 而不是检查字节，因此 Avro 到 Iceberg 的类型映射是精确的：

GSR 集成透明地处理架构版本控制。一旦生产者注册了包含额外字段的新架构版本，Flink 消费者将反序列化更新后的负载并调整 Iceberg 表以匹配，无需重启作业。

先决条件

要遵循本指南，您需要以下内容：

• 具有创建 Amazon Kinesis Data Streams、Managed Service for Apache Flink 应用程序、AWS Glue 资源和 Amazon S3 存储桶权限的 AWS 账户（如果选择该目录，还需要 Amazon S3 Tables）。
• 配置了凭据的 AWS 命令行接口（AWS CLI）。
• Node.js 18 或更高版本，以及 AWS Cloud Development Kit（AWS CDK）CLI。
• Java 17 或更高版本以及 Apache Maven 3.9 或更高版本，用于构建数据生成器。
• 在本地运行 Docker。CDK 构建在 Maven 镜像内捆绑 Flink 应用程序 jar。

部署并测试解决方案

随附的存储库通过单个参数化的 AWS CDK 堆栈预置所有内容。

cd cdk-infrastructure && npm install
npx cdk bootstrap aws://<account>/<region>

npx cdk deploy -c appType=dynamic -c tableFormatVersion=2 # JSON inference variant
npx cdk deploy -c appType=dynamic-avro -c tableFormatVersion=2 # GSR Avro variant

aws kinesisanalyticsv2 start-application --application-name <ApplicationName> --run-configuration 'ApplicationRestoreConfiguration={ApplicationRestoreType=SKIP_RESTORE_FROM_SNAPSHOT}'

java -jar data-generator/target/data-generator-1.0-SNAPSHOT.jar <stream-name> <region> 100 60 v1

java -jar data-generator/target/data-generator-1.0-SNAPSHOT.jar <stream-name> <region> 100 60 v2

java -jar data-generator/target/data-generator-1.0-SNAPSHOT.jar avro <stream-name> <region> <registry-name> 100 60

1. 安装 CDK 依赖项并引导您的环境（仅首次）：
2. 部署您想要尝试的变体： 向任一命令添加 -c catalogType=s3tables 以使用 Amazon S3 Tables 而不是 AWS Glue Data Catalog。演练设置 tableFormatVersion=2，以便您可以使用广泛的引擎查询结果。当您使用支持 v3 的引擎（如 Spark on Amazon EMR 7.12+ 或 AWS Glue ETL）查询时，可省略它以使用默认的 Iceberg 格式版本 3。
3. 使用堆栈输出中的 ApplicationName 值启动应用程序：
4. 使用附带的数据生成器发送测试事件。从 v1 负载开始，这些负载创建的表不包含可选字段： 然后发送添加 userAgent 和 scrollDepth 字段的 v2 负载。这第二次运行是您在下一步中观察到的架构演化： 对于 Avro 变体，生成器在发送时将每个架构版本注册到 AWS Glue Schema Registry 中：
5. 在 Amazon Athena 中查询路由表。您应该会在一个检查点间隔内看到每个事件类型一个 Iceberg 表出现在数据库中，并且在发送 v2 事件后，新字段（userAgent、scrollDepth）显示为同一表上的可选列。Iceberg 元数据表（例如，SELECT * FROM "db"."table$snapshots"）显示 sink 所做的每次提交。

清理

完成测试后，删除资源以避免产生费用：

cd cdk-infrastructure && npx cdk destroy

CDK会删除Kinesis数据流、Apache Flink的托管服务应用程序以及由堆栈创建的AWS身份与访问管理（IAM）角色。此外，请清空并删除S3数据仓库存储桶以移除Iceberg数据和元数据文件，删除Avro变体在AWS Glue Schema Registry中注册的任何模式，以及如果您使用了S3 Tables目录，请删除表存储桶中的内容。

结论

利用Apache Iceberg 1.11.0和Flink 2.3，您可以构建能够适应变化而无需停止管道的流式数据湖架构。通过逐条记录路由，单个Flink应用程序可以将多种事件类型写入不同的Iceberg表，而自动模式演进则使表定义与不断变化的源数据保持一致。选择AWS Glue Schema Registry而非运行时JSON推断，可提供精确的类型和受控的演进契约，而可配置的分区候选列表可确保每个路由的表在无需预先声明其模式的情况下正确分区。

结果是减少管道重新部署次数、降低运营开销，并使数据湖与不断演进的应用程序模式保持同步。

要开始使用，请按照“部署与测试”部分操作，然后根据您自己的事件类型调整路由字段和分区候选。

完整的示例代码可在随附的GitHub存储库中获取。