如何在区域AWS Lambda中订阅DynamoDB全局表流？

使用CDK部署Lambda堆栈时，因DynamoDB全局表的流ARN配置错误导致部署失败。代码中使用stream/latest作为流ARN的一部分，但Lambda服务无法识别该占位符，报错提示流不存在。

相关代码：

const globalTableTableName = "my-org-glb-table1"

const regionalTable = Table.fromTableAttributes(this, `Table-${Stack.of(this).region}`, {
    tableName: globalTableTableName,
    tableStreamArn: `arn:aws:dynamodb:${Stack.of(this).region}:${this.account}:table/${globalTableTableName}/stream/latest` // 此处存在问题！
});

const eventSource = new DynamoEventSource(regionalTable, {
    startingPosition: StartingPosition.TRIM_HORIZON,
    batchSize: 5,
    bisectBatchOnError: true,
    retryAttempts: 2
});
lambda1.addEventSource(eventSource);

报错信息：

❌ 部署失败：Error: 名为app-my-org-lambda-stack的堆栈创建失败，可能需要从AWS控制台手动删除：ROLLBACK_COMPLETE: Resource handler returned message: "Invalid request provided: Stream not found: arn:aws:dynamodb:us-east-1:123456789012:table/app-my-org-glb-table1/stream/latest (Service: Lambda, Status Code: 400, Request ID: e165f9e7-f808-4528-b383-4992861e5aa0)" (RequestToken: d98d6f65-21d1-80b0-4ab3-7a3fc9430268, HandlerErrorCode: InvalidRequest)

补充说明：已通过L1构造CfnGlobalTable创建全局表，且已验证表可部署并开启了流（带有特定ARN）。

stream/latest是AWS控制台提供的快捷方式，用于动态指向当前最新的流，但Lambda等AWS服务不支持将其作为有效的ARN部分使用。实际的DynamoDB流ARN包含唯一的流ID（格式类似2024-05-20T12:34:56.789），必须使用这个具体的ARN才能让Lambda正确识别流资源。

步骤1：确保全局表每个副本都开启流并配置输出 #

在创建全局表的主区域堆栈中，为每个区域的副本配置流规格，并将对应区域的流ARN作为堆栈输出导出：

const PRIMARY_REGION = 'us-east-1';
const SECONDARY_REGIONS = ['us-west-2', 'eu-west-1']; // 根据实际需求添加区域

const globalTable = new CfnGlobalTable(this, 'my-org-glb-table1', {
  tableName: "my-org-glb-table1",
  keySchema: [
    { attributeName: 'id', keyType: 'HASH' },
    // 其他键定义
  ],
  attributeDefinitions: [
    { attributeName: 'id', attributeType: 'S' },
    // 其他属性定义
  ],
  replicas: [
    {
      region: PRIMARY_REGION,
      streamSpecification: {
        streamViewType: 'NEW_AND_OLD_IMAGES' // 根据业务需求选择流视图类型
      }
    },
    ...SECONDARY_REGIONS.map(region => ({
      region,
      streamSpecification: {
        streamViewType: 'NEW_AND_OLD_IMAGES'
      }
    }))
  ],
  // 其他必要属性，如 billingMode 等
});

// 导出主区域的流ARN
new CfnOutput(this, 'PrimaryStreamArn', {
  value: globalTable.getAtt('Replicas.0.StreamArn').toString(),
  exportName: `my-org-glb-table1-stream-arn-${PRIMARY_REGION}`
});

// 导出次级区域的流ARN
SECONDARY_REGIONS.forEach((region, index) => {
  new CfnOutput(this, `${region}StreamArn`, {
    value: globalTable.getAtt(`Replicas.${index + 1}.StreamArn`).toString(),
    exportName: `my-org-glb-table1-stream-arn-${region}`
  });
});

步骤2：在Lambda区域堆栈中导入对应流ARN #

在Lambda所在的区域堆栈中，通过Fn.importValue导入对应区域的流ARN，替换硬编码的stream/latest：

const globalTableTableName = "my-org-glb-table1";
const currentRegion = Stack.of(this).region;

// 导入当前区域的流ARN
const streamArn = Fn.importValue(`my-org-glb-table1-stream-arn-${currentRegion}`);

const regionalTable = Table.fromTableAttributes(this, `Table-${currentRegion}`, {
  tableName: globalTableTableName,
  tableStreamArn: streamArn
});

const eventSource = new DynamoEventSource(regionalTable, {
  startingPosition: StartingPosition.TRIM_HORIZON,
  batchSize: 5,
  bisectBatchOnError: true,
  retryAttempts: 2
});
lambda1.addEventSource(eventSource);

额外说明 #

• 如果全局表后续更新了流（比如重新开启流），对应的流ARN会变化，此时需要重新部署全局表堆栈以更新输出，再重新部署Lambda堆栈导入新的ARN。
• 流视图类型（streamViewType）可根据业务需求选择KEYS_ONLY、NEW_IMAGE、OLD_IMAGE或NEW_AND_OLD_IMAGES。

内容的提问来源于stack exchange，提问作者ses