新架構設計

Lambda 程式碼：boto3 dynamodb → boto3 rds-data

這是整個替換中，程式碼層面最核心的改變，把原本 DynamoDB 的部分換成讀 Aurora PostgreSQL 。

import boto3
import os

rds = boto3.client('rds-data')
CLUSTER = os.environ['DB_CLUSTER_ARN']
SECRET = os.environ['DB_SECRET_ARN']
DB = os.environ['DB_NAME']


def _exec(sql, params=None):
    kwargs = dict(resourceArn=CLUSTER, secretArn=SECRET, database=DB, sql=sql)
    if params:
        kwargs['parameters'] = params
    return rds.execute_statement(**kwargs)


# 寫入（Parameterized query 防止 SQL injection）
_exec(
    """
    INSERT INTO ad_clicks
        (click_id, ad_id, impression_id, user_id, ts, ts_iso, source_url, processed_at, kinesis_seq_no)
    VALUES
        (:click_id, :ad_id, :impression_id, :user_id, :ts, :ts_iso, :source_url, :processed_at, :kinesis_seq_no)
    """,
    [
        {'name': 'click_id', 'value': {'stringValue': click_id}},
        {'name': 'ad_id', 'value': {'stringValue': ad_id}},
        {'name': 'ts', 'value': {'longValue': ts}},
        # … 其餘欄位
    ],
)

# 查詢（支援複雜條件）
resp = rds.execute_statement(
    resourceArn=CLUSTER,
    secretArn=SECRET,
    database=DB,
    sql='SELECT * FROM ad_clicks WHERE ad_id = :ad_id ORDER BY ts DESC',
    parameters=[{'name': 'ad_id', 'value': {'stringValue': ad_id}}],
    includeResultMetadata=True,
)

# 將結果轉換成 dict list
cols = [c['label'] for c in resp['columnMetadata']]
items = [
    dict(zip(cols, [list(v.values())[0] for v in row]))
    for row in resp['records']
]

驗證：end-to-end 測試

Deploy 完成後，可以直接打 POST API 把資料送進 PostgreSQL 。

curl -X POST https://<api-id>.execute-api.us-east-1.amazonaws.com/dev/clicks \
-H "Content-Type: application/json" \
-d '{"ad_id":"ad_001","impression_id":"imp_001","user_id":"user_A","source_url":"https://example.com"}'

等個幾秒之後，在打查詢 API 確認。

curl https://<api-id>.execute-api.us-east-1.amazonaws.com/dev/stats/ad_001

回傳結果。

{
  "ad_id": "ad_001",
  "total_clicks": 3,
  "unique_users": 2,
  "unique_impressions": 3,
  "clicks_by_date": {
    "2026-05-30": 3
  },
  "recent_clicks": [
    {
      "click_id": "b31d24fa-797b-4b57-a9bf-08c4cde5eed6",
      "impression_id": "imp_002",
      "user_id": "user_B",
      "timestamp": 1780147043,
      "source_url": "https://example.com/page2"
    }
    ...
  ]
}

Medium: 使用 Serverless 架構設計廣告點擊系統 — 把 API 的 Data Source 從 DynamoDB 換成 Aurora PostgreSQL