加入 S3 延伸廣告點擊系統架構

為了在後續做分析時可以節省更多成本，在原本的架構上加入 S3 存放資料，不只省錢，也可以當作備份。

CloudFormation的部分，可以多創建一個 S3 Bucket

Resources:
  # S3 Bucket for storing raw ad click events
  AdClickBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub 'ad-clicks-raw-${Environment}-${AWS::AccountId}'
      LifecycleConfiguration:
        Rules:
          - Id: TransitionToIA
            Status: Enabled
            Transitions:
              - TransitionInDays: 30
                StorageClass: STANDARD_IA
          - Id: TransitionToGlacier
            Status: Enabled
            Transitions:
              - TransitionInDays: 90
                StorageClass: GLACIER
      VersioningConfiguration:
        Status: Enabled
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      Tags:
        - Key: Environment
          Value: !Ref Environment
        - Key: Application
          Value: AdClickAggregator

如果需要限制存取，可以在 bucket 設定 policy。

# S3 Bucket Policy
  AdClickBucketPolicy:
    Type: AWS::S3::BucketPolicy
    Properties:
      Bucket: !Ref AdClickBucket
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: DenyInsecureConnections
            Effect: Deny
            Principal: '*'
            Action: 's3:*'
            Resource:
              - !GetAtt AdClickBucket.Arn
              - !Sub '${AdClickBucket.Arn}/*'
            Condition:
              Bool:
                'aws:SecureTransport': false

在 Lambda 的 role 上，也需要加上 policy 允許 Lambda 可以把資料 put 上去 bucket 。

# IAM Role for Lambda Function
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub 'ad-click-lambda-role-${Environment}'
      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
      Policies:
        - PolicyName: DynamoDBAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - dynamodb:PutItem
                  - dynamodb:UpdateItem
                  - dynamodb:GetItem
                  - dynamodb:Query
                  - dynamodb:Scan
                Resource: !GetAtt AdClickTable.Arn
        - PolicyName: S3Access
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - s3:PutObject
                  - s3:PutObjectAcl
                  - s3:GetObject
                Resource: !Sub '${AdClickBucket.Arn}/*'
        - PolicyName: SQSAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - sqs:ReceiveMessage
                  - sqs:DeleteMessage
                  - sqs:GetQueueAttributes
                Resource: !GetAtt AdClickQueue.Arn

程式碼的部分，除了把資料新增 DynamoDB 之外，也需要加上上傳到 S3 的邏輯。

# Lambda Function for processing ad clicks
  AdClickProcessorFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub 'ad-click-processor-${Environment}'
      Runtime: python3.11
      Handler: index.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Timeout: 60
      MemorySize: 256
      Environment:
        Variables:
          DYNAMODB_TABLE: !Ref AdClickTable
          S3_BUCKET: !Ref AdClickBucket
          ENVIRONMENT: !Ref Environment
      Code:
        ZipFile: |
          import json
          import boto3
          import os
          import uuid
          from datetime import datetime
          from decimal import Decimal

          dynamodb = boto3.resource('dynamodb')
          s3_client = boto3.client('s3')
          table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])
          s3_bucket = os.environ['S3_BUCKET']

          def lambda_handler(event, context):
              print(f"Processing {len(event['Records'])} records")
              
              processed = 0
              failed = 0
              
              for record in event['Records']:
                  try:
                      body = json.loads(record['body'])
                      print(f"Received body: {json.dumps(body)}")
                      
                      ad_id = body.get('ad_id')
                      impression_id = body.get('impression_id')
                      user_id = body.get('user_id', 'anonymous')
                      timestamp_str = body.get('timestamp')
                      source_url = body.get('source_url')
                      
                      if not ad_id:
                          raise ValueError("ad_id is required")
                      if not impression_id:
                          raise ValueError("impression_id is required")
                      
                      click_id = str(uuid.uuid4())
                      
                      if timestamp_str:
                          dt = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
                          timestamp = int(dt.timestamp())
                      else:
                          dt = datetime.now()
                          timestamp = int(dt.timestamp())
                          timestamp_str = dt.isoformat()
                      
                      processed_at = datetime.now().isoformat()
                      
                      # Prepare item for DynamoDB
                      item = {
                          'ad_id': ad_id,
                          'click_id': click_id,
                          'impression_id': impression_id,
                          'user_id': user_id,
                          'timestamp': timestamp,
                          'timestamp_iso': timestamp_str,
                          'source_url': source_url,
                          'processedAt': processed_at
                      }
                      
                      # Write to DynamoDB
                      table.put_item(Item=item)
                      print(f"Wrote to DynamoDB: {ad_id}/{click_id}")
                      
                      # Write to S3 (partitioned by date and ad_id)
                      date_partition = datetime.fromtimestamp(timestamp).strftime('%Y/%m/%d')
                      s3_key = f"raw-clicks/{date_partition}/ad_id={ad_id}/{click_id}.json"
                      
                      # Prepare S3 object (convert Decimal to int/float for JSON serialization)
                      s3_object = {
                          'ad_id': ad_id,
                          'click_id': click_id,
                          'impression_id': impression_id,
                          'user_id': user_id,
                          'timestamp': timestamp,
                          'timestamp_iso': timestamp_str,
                          'source_url': source_url,
                          'processedAt': processed_at,
                          'sqs_message_id': record.get('messageId')
                      }
                      
                      s3_client.put_object(
                          Bucket=s3_bucket,
                          Key=s3_key,
                          Body=json.dumps(s3_object, indent=2),
                          ContentType='application/json',
                          Metadata={
                              'ad_id': ad_id,
                              'click_id': click_id,
                              'user_id': user_id
                          }
                      )
                      print(f"Wrote to S3: s3://{s3_bucket}/{s3_key}")
                      
                      processed += 1
                      print(f"Successfully processed click for ad: {ad_id}, impression: {impression_id}")
                      
                  except Exception as e:
                      failed += 1
                      print(f"Error processing record: {str(e)}")
                      print(f"Record: {record}")
                      # Don't raise exception to continue processing other records
              
              return {
                  'statusCode': 200,
                  'body': json.dumps({
                      'processed': processed,
                      'failed': failed,
                      'total': len(event['Records'])
                  })
              }
      Tags:
        - Key: Environment
          Value: !Ref Environment
        - Key: Application
          Value: AdClickAggregator

這樣就完成了，打 API 後就可以看到資料上傳到 S3 bucket 了。

json 檔裡面的數據會長的像下面這樣。

{
  "ad_id": "ad_1",
  "click_id": "0001fdd5-0ea5-44b7-ac39-3567d2287bfe",
  "impression_id": "imp_66",
  "user_id": "user_88",
  "timestamp": 1764799859,
  "timestamp_iso": "2025-12-03T22:10:59Z",
  "source_url": "https://gtntjplb.example.com/",
  "processedAt": "2025-12-03T14:11:00.760843",
  "sqs_message_id": "3798b960-d129-4d37-89be-50cd2656a17e"
}