AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Description: EventBridge Pipe that connects DynamoDB Streams to Lambda with Enrichment Resources: # DynamoDB Table Creation with Stream Enabled DynamoDBSampleTable: Type: AWS::DynamoDB::Table Properties: TableName: SampleTable AttributeDefinitions: - AttributeName: PK AttributeType: S - AttributeName: SK AttributeType: S KeySchema: - AttributeName: PK KeyType: HASH - AttributeName: SK KeyType: RANGE ProvisionedThroughput: ReadCapacityUnits: 5 WriteCapacityUnits: 5 StreamSpecification: StreamViewType: NEW_IMAGE # Pipes target (Lambda function) Target: Type: AWS::Serverless::Function Properties: FunctionName: !Sub ${AWS::StackName}-target-lambda CodeUri: src/ Handler: lambda_function.lambda_handler Runtime: python3.9 MemorySize: 128 Timeout: 15 Architectures: - arm64 # Enrichment log group EnrichmentLogGroup: Type: AWS::Logs::LogGroup Properties: LogGroupName: !Join ['/', ['stepfunctions', StateMachine]] # Enrichment with Step Functions EnrichmentStateMachine: Type: AWS::Serverless::StateMachine Properties: Type: EXPRESS DefinitionUri: workflow/stepfunction.asl.json Role: !GetAtt EnrichmentStateMachineRole.Arn Logging: Destinations: - CloudWatchLogsLogGroup: LogGroupArn: !GetAtt EnrichmentLogGroup.Arn IncludeExecutionData: true Level: 'ALL' #Execution Role for StepFunctions EnrichmentStateMachineRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Principal: Service: - states.amazonaws.com Action: - sts:AssumeRole Policies: - PolicyName: CloudWatchLogs PolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Action: - 'logs:CreateLogDelivery' - 'logs:GetLogDelivery' - 'logs:UpdateLogDelivery' - 'logs:DeleteLogDelivery' - 'logs:ListLogDeliveries' - 'logs:PutResourcePolicy' - 'logs:DescribeResourcePolicies' - 'logs:DescribeLogGroups' Resource: '*' # Role + Permissions for Pipes PipeRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Principal: Service: - pipes.amazonaws.com Action: - sts:AssumeRole Policies: - PolicyName: SourcePolicy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - "dynamodb:DescribeStream" - "dynamodb:GetRecords" - "dynamodb:GetShardIterator" - "dynamodb:ListStreams" Resource: !GetAtt DynamoDBSampleTable.StreamArn - PolicyName: TargetPolicy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - 'lambda:InvokeFunction' Resource: !GetAtt Target.Arn - PolicyName: EnrichmentPolicy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - 'states:StartExecution' - 'states:StartSyncExecution' Resource: !GetAtt EnrichmentStateMachine.Arn Pipe: Type: AWS::Pipes::Pipe Properties: Name: ddb-to-lambda-enrichment Description: "Pipe to connect DynamoDB Stream to AWS Lambda with filtering and enrichment" RoleArn: !GetAtt PipeRole.Arn Source: !GetAtt DynamoDBSampleTable.StreamArn SourceParameters: FilterCriteria: Filters: - Pattern: '{ "eventName": ["INSERT"], "dynamodb": { "NewImage": { "messageId": { "S": [{ "exists": true }] }, "PK": { "S": [{ "prefix": "Message#" }] }, "SK": { "S": [{ "prefix": "Channel#" }] } } } }' DynamoDBStreamParameters: StartingPosition: LATEST BatchSize: 1 Enrichment: !GetAtt EnrichmentStateMachine.Arn EnrichmentParameters: InputTemplate: '{ "PK": <$.dynamodb.NewImage.PK.S>, "SK": <$.dynamodb.NewImage.SK.S> }' Target: !GetAtt Target.Arn