TheArchitect

Data Pipeline

AWS architecture template · S3 trigger → Lambda → DynamoDB + SQS + CloudWatch

Data Pipeline is a free AWS architecture template built with CloudWatch, DynamoDB, IAM, Lambda, S3 and SQS. Open it in TheArchitect to edit the diagram, estimate monthly cost, and export the CloudFormation or Terraform that deploys it.

Open in TheArchitect All templates
Ingest BucketWorker LambdaWorker RolePipeline TablePipeline QueueWorker Alarm
Data Pipeline architecture diagram

AWS services in this template

  • CloudWatch
  • DynamoDB
  • IAM
  • Lambda
  • S3
  • SQS

Deployment notes

Deploy-ready

  • Includes a private encrypted bucket, inline worker code, execution role, queue, table, and alarm settings.

Generated CloudFormation — 8 resources

Logical IDResource type
IngestBuckets3AWS::S3::Bucket
WorkerLambdalambdaAWS::Lambda::Function
WorkerRoleiamAWS::IAM::Role
PipelineTabledynamodbAWS::DynamoDB::Table
PipelineQueuesqsAWS::SQS::Queue
WorkerAlarmcloudwatchAWS::CloudWatch::Alarm
WorkerLambdalambdaIngestBuckets3PermissionAWS::Lambda::Permission
WorkerLambdalambdaPipelineQueuesqsEventSourceMappingAWS::Lambda::EventSourceMapping
View the full CloudFormation template
{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Description": "Generated by TheArchitect — https://www.thecloudarchitect.dev",
  "Metadata": {
    "TheArchitect": {
      "Edges": [
        {
          "From": "IngestBuckets3",
          "To": "WorkerLambdalambda",
          "Label": "trigger"
        },
        {
          "From": "WorkerRoleiam",
          "To": "WorkerLambdalambda",
          "Label": "role"
        },
        {
          "From": "WorkerLambdalambda",
          "To": "PipelineTabledynamodb",
          "Label": "write"
        },
        {
          "From": "WorkerLambdalambda",
          "To": "PipelineQueuesqs",
          "Label": "queue"
        },
        {
          "From": "WorkerLambdalambda",
          "To": "WorkerAlarmcloudwatch",
          "Label": "logs"
        }
      ]
    }
  },
  "Resources": {
    "IngestBuckets3": {
      "Type": "AWS::S3::Bucket",
      "Properties": {
        "BucketName": "architect-demo-ingest",
        "VersioningConfiguration": {
          "Status": "Enabled"
        },
        "OwnershipControls": {
          "Rules": [
            {
              "ObjectOwnership": "BucketOwnerEnforced"
            }
          ]
        },
        "BucketEncryption": {
          "ServerSideEncryptionConfiguration": [
            {
              "ServerSideEncryptionByDefault": {
                "SSEAlgorithm": "AES256"
              }
            }
          ]
        },
        "PublicAccessBlockConfiguration": {
          "BlockPublicAcls": true,
          "BlockPublicPolicy": true,
          "IgnorePublicAcls": true,
          "RestrictPublicBuckets": true
        },
        "Tags": [
          {
            "Key": "Name",
            "Value": "architect-demo-ingest"
          }
        ],
        "NotificationConfiguration": {
          "LambdaConfigurations": [
            {
              "Event": "s3:ObjectCreated:*",
              "Function": {
                "Fn::GetAtt": [
                  "WorkerLambdalambda",
                  "Arn"
                ]
              }
            }
          ]
        }
      },
      "DependsOn": [
        "WorkerLambdalambdaIngestBuckets3Permission"
      ]
    },
    "WorkerLambdalambda": {
      "Type": "AWS::Lambda::Function",
      "DependsOn": [
        "WorkerRoleiam"
      ],
      "Properties": {
        "FunctionName": "architect-demo-pipeline-worker",
        "Runtime": "nodejs20.x",
        "Handler": "index.handler",
        "MemorySize": 512,
        "Architectures": [
          "arm64"
        ],
        "Timeout": 30,
        "Description": "Processes object uploads and sends normalized events downstream.",
        "Environment": {
          "Variables": {
            "APP_ENV": "demo",
            "LOG_LEVEL": "info"
          }
        },
        "TracingConfig": {
          "Mode": "Active"
        },
        "Code": {
          "ZipFile": "exports.handler = async (event) => ({ processed: Array.isArray(event.Records) ? event.Records.length : 1 });"
        },
        "Role": {
          "Fn::GetAtt": [
            "WorkerRoleiam",
            "Arn"
          ]
        }
      }
    },
    "WorkerRoleiam": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "RoleName": "architect-demo-pipeline-worker-role",
        "Description": "Execution role for the data pipeline worker.",
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": {
                "Service": "lambda.amazonaws.com"
              },
              "Action": "sts:AssumeRole"
            }
          ]
        },
        "ManagedPolicyArns": [
          "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
          "arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole"
        ]
      }
    },
    "PipelineTabledynamodb": {
      "Type": "AWS::DynamoDB::Table",
      "Properties": {
        "TableName": "architect-demo-pipeline-items",
        "BillingMode": "PAY_PER_REQUEST",
        "AttributeDefinitions": [
          {
            "AttributeName": "pk",
            "AttributeType": "S"
          },
          {
            "AttributeName": "sk",
            "AttributeType": "S"
          }
        ],
        "KeySchema": [
          {
            "AttributeName": "pk",
            "KeyType": "HASH"
          },
          {
            "AttributeName": "sk",
            "KeyType": "RANGE"
          }
        ],
        "PointInTimeRecoverySpecification": {
          "PointInTimeRecoveryEnabled": true
        },
        "SSESpecification": {
          "SSEEnabled": true
        }
      }
    },
    "PipelineQueuesqs": {
      "Type": "AWS::SQS::Queue",
      "Properties": {
        "QueueName": "architect-demo-pipeline-dlq",
        "VisibilityTimeout": 120,
        "MessageRetentionPeriod": 1209600,
        "ReceiveMessageWaitTimeSeconds": 20
      }
    },
    "WorkerAlarmcloudwatch": {
      "Type": "AWS::CloudWatch::Alarm",
      "Properties": {
        "AlarmName": "architect-demo-pipeline-errors",
        "AlarmDescription": "Demo alarm for pipeline worker errors.",
        "Namespace": "AWS/Lambda",
        "MetricName": "Errors",
        "Statistic": "Sum",
        "Period": 60,
        "EvaluationPeriods": 1,
        "Threshold": 1,
        "ComparisonOperator": "GreaterThanOrEqualToThreshold",
        "TreatMissingData": "notBreaching"
      }
    },
    "WorkerLambdalambdaIngestBuckets3Permission": {
      "Type": "AWS::Lambda::Permission",
      "DependsOn": [
        "WorkerLambdalambda"
      ],
      "Properties": {
        "Action": "lambda:InvokeFunction",
        "FunctionName": {
          "Ref": "WorkerLambdalambda"
        },
        "Principal": "s3.amazonaws.com",
        "SourceAccount": {
          "Ref": "AWS::AccountId"
        },
        "SourceArn": {
          "Fn::Sub": "arn:${AWS::Partition}:s3:::architect-demo-ingest"
        }
      }
    },
    "WorkerLambdalambdaPipelineQueuesqsEventSourceMapping": {
      "Type": "AWS::Lambda::EventSourceMapping",
      "DependsOn": [
        "WorkerLambdalambda",
        "PipelineQueuesqs"
      ],
      "Properties": {
        "EventSourceArn": {
          "Fn::GetAtt": [
            "PipelineQueuesqs",
            "Arn"
          ]
        },
        "FunctionName": {
          "Ref": "WorkerLambdalambda"
        },
        "Enabled": true,
        "BatchSize": 10
      }
    }
  }
}

Generated by TheArchitect from the diagram above. Review values such as names, CIDR ranges and instance sizes before deploying to your account.

Related templates