求CDK TypeScript实现Lambda接收DynamoDB流并发布到SNS主题的示例

前置依赖 #

你需要提前安装对应的CDK依赖包：

npm install aws-cdk-lib @aws-sdk/client-sns @aws-sdk/util-dynamodb constructs

1. CDK 栈代码 #

import * as cdk from 'aws-cdk-lib';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as lambdaEventSources from 'aws-cdk-lib/aws-lambda-event-sources';
import * as sns from 'aws-cdk-lib/aws-sns';
import { Construct } from 'constructs';

export class DynamoStreamToSnsStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // 1. 创建启用流的DynamoDB表
    const demoTable = new dynamodb.Table(this, 'DemoTable', {
      partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING },
      stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES, // 按需调整流返回内容
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 测试环境用，生产环境可删除
    });

    // 2. 创建SNS主题
    const changeNotifyTopic = new sns.Topic(this, 'ChangeNotifyTopic', {
      displayName: 'DynamoDB变更通知主题',
    });

    // 3. 创建处理DynamoDB流的Lambda函数
    const streamHandler = new lambda.NodejsFunction(this, 'StreamHandler', {
      runtime: lambda.Runtime.NODEJS_20_X,
      entry: 'lambda/stream-handler.ts', // 替换为你的Lambda代码实际路径
      environment: {
        SNS_TOPIC_ARN: changeNotifyTopic.topicArn, // 把SNS主题ARN注入Lambda环境变量
      },
    });

    // 核心配置1：给Lambda授予SNS主题发布权限（CDK自动处理IAM规则，无需手动写策略）
    changeNotifyTopic.grantPublish(streamHandler);

    // 核心配置2：给Lambda添加DynamoDB流触发源，自动授予流读取权限
    streamHandler.addEventSource(new lambdaEventSources.DynamoEventSource(demoTable, {
      startingPosition: lambda.StartingPosition.LATEST,
      batchSize: 10, // 按需调整批量处理大小
      reportBatchItemFailures: true, // 开启部分失败重试配置
    }));
  }
}

2. Lambda 业务代码（对应路径lambda/stream-handler.ts） #

import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
import { unmarshall } from "@aws-sdk/util-dynamodb";
import type { DynamoDBStreamEvent } from "aws-lambda";

const snsClient = new SNSClient({});
const snsTopicArn = process.env.SNS_TOPIC_ARN as string;

export const handler = async (event: DynamoDBStreamEvent) => {
  for (const record of event.Records) {
    if (!record.dynamodb?.NewImage) continue;

    // 把DynamoDB带类型的记录转成普通JS对象
    const changeData = unmarshall(record.dynamodb.NewImage);
    // 如需获取变更前旧数据可使用：record.dynamodb.OldImage ? unmarshall(record.dynamodb.OldImage) : null
    const operationType = record.eventName; // 取值为INSERT/MODIFY/REMOVE

    // 发布消息到SNS
    await snsClient.send(new PublishCommand({
      TopicArn: snsTopicArn,
      Message: JSON.stringify({
        operationType,
        changeData,
        eventSource: record.eventSourceARN,
        timestamp: Date.now()
      }),
      Subject: `DynamoDB${operationType}变更通知`
    }));
  }

  return { statusCode: 200, body: '处理完成' };
};

关键说明 #

• 权限配置是CDK侧实现Lambda发SNS的核心，直接调用SNS主题实例的grantPublish方法传入Lambda实例即可，CDK会自动生成对应的IAM策略，不需要手动拼接ARN和权限项
• DynamoDB流的视图类型可以按需调整，如果你只需要变更后的新数据，设置为StreamViewType.NEW_IMAGE即可
• Node.js 18.x及以上的Lambda运行环境已经内置AWS SDK v3，不需要把SDK打包到Lambda部署包中，可以减小包体积

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