因為之前都是把資料存到 DynamoDB ，為了把之前的舊資料遷移 PostgreSQL，可以先把 S3 的資料格式轉成 Parquet ，在用 EMR 把資料 migrate 到 Aurora PostgreSQL，這樣就可以讓 API 改從 PostgreSQL 查詢資料。
這整段遷移的流程，可以使用 CloudFormation 把 EMR Serverless 的 sinfra 、相關的 role 和權限設定好，就可以跑 script 觸發 migrate 的流程。
AWSTemplateFormatVersion: "2010-09-09"
Description: "EMR Serverless application, IAM roles, and S3 artifacts bucket"
Parameters:
StackName:
Type: String
PrivateSubnet1Id:
Type: AWS::EC2::Subnet::Id
PrivateSubnet2Id:
Type: AWS::EC2::Subnet::Id
EmrSecurityGroupId:
Type: AWS::EC2::SecurityGroup::Id
SecretArn:
Type: String
Description: Secrets Manager ARN for Aurora credentials
Resources:
# ── S3 Artifacts Bucket (PySpark script + JDBC JAR + logs) ────────────────
ArtifactsBucket:
Type: AWS::S3::Bucket
DeletionPolicy: Retain
UpdateReplacePolicy: Retain
Properties:
BucketName: !Sub "${StackName}-artifacts-${AWS::AccountId}"
VersioningConfiguration:
Status: Enabled
Tags:
- Key: Name
Value: !Sub "${StackName}-artifacts"
# ── IAM Role for EMR Serverless Job Execution ─────────────────────────────
EmrJobRole:
Type: AWS::IAM::Role
Properties:
RoleName: !Sub "${StackName}-emr-job-role"
AssumeRolePolicyDocument:
Version: "2012-10-17"
Statement:
- Effect: Allow
Principal:
Service: emr-serverless.amazonaws.com
Action: "sts:AssumeRole"
Policies:
- PolicyName: EmrJobPolicy
PolicyDocument:
Version: "2012-10-17"
Statement:
# Read source Parquet files from S3
- Effect: Allow
Action:
- s3:GetObject
- s3:ListBucket
Resource:
- "arn:aws:s3:::ad-clicks-raw-dev-262969866776"
- "arn:aws:s3:::ad-clicks-raw-dev-262969866776/*"
# Read/write artifacts bucket (script, JDBC JAR, logs)
- Effect: Allow
Action:
- s3:GetObject
- s3:PutObject
- s3:ListBucket
- s3:DeleteObject
Resource:
- !GetAtt ArtifactsBucket.Arn
- !Sub "${ArtifactsBucket.Arn}/*"
# Read Aurora credentials from Secrets Manager
- Effect: Allow
Action:
- secretsmanager:GetSecretValue
Resource: !Ref SecretArn
# Write CloudWatch logs
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
- logs:DescribeLogGroups
- logs:DescribeLogStreams
Resource: "*"
# ── EMR Serverless Application ─────────────────────────────────────────────
EmrServerlessApplication:
Type: AWS::EMRServerless::Application
Properties:
Name: !Sub "${StackName}-spark"
ReleaseLabel: "emr-6.15.0"
Type: SPARK
NetworkConfiguration:
SubnetIds:
- !Ref PrivateSubnet1Id
- !Ref PrivateSubnet2Id
SecurityGroupIds:
- !Ref EmrSecurityGroupId
AutoStartConfiguration:
Enabled: true
AutoStopConfiguration:
Enabled: true
IdleTimeoutMinutes: 15
Tags:
- Key: Name
Value: !Sub "${StackName}-spark"
Outputs:
EmrApplicationId:
Value: !Ref EmrServerlessApplication
Description: EMR Serverless Application ID
EmrJobRoleArn:
Value: !GetAtt EmrJobRole.Arn
Description: IAM Role ARN for EMR job execution
ArtifactsBucket:
Value: !Ref ArtifactsBucket
Description: S3 bucket for job artifacts (scripts, JARs, logs)
EMR（Elastic MapReduce）是 AWS 提供處理大數據的框架，讓資料工程師可以快速的處理資料，只要：
"""
PySpark job: S3 Parquet → Aurora PostgreSQL one-time migration.
Environment variables (injected by EMR Serverless job configuration):
S3_SOURCE_PATH - s3://bucket/parquet-clicks/
AURORA_ENDPOINT - Aurora cluster writer endpoint hostname
SECRET_ARN - Secrets Manager ARN for DB credentials
DB_NAME - Aurora database name (default: adclicks)
"""
import json
import logging
import os
import re
import time
from typing import Optional
import boto3
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
logger = logging.getLogger(__name__)
# ── Regex patterns ────────────────────────────────────────────────────────────
_PATH_PATTERN = re.compile(r"parquet-clicks/(\d{4})/(\d{2})/(\d{2})/")
_AD_ID_PATTERN = re.compile(r"ad_id=([^/]+)/")
# ── T013: Path parsing helpers (pure Python — unit-testable without Spark) ────
def extract_year(path: str) -> Optional[int]:
m = _PATH_PATTERN.search(path)
return int(m.group(1)) if m else None
def extract_month(path: str) -> Optional[int]:
m = _PATH_PATTERN.search(path)
return int(m.group(2)) if m else None
def extract_day(path: str) -> Optional[int]:
m = _PATH_PATTERN.search(path)
return int(m.group(3)) if m else None
# ── T015: ad_id path extraction + mismatch detection ─────────────────────────
def extract_path_ad_id(path: str) -> Optional[str]:
m = _AD_ID_PATTERN.search(path)
return m.group(1) if m else None
def find_ad_id_mismatches(rows: list) -> list:
"""
Given a list of dicts with 'file_path' and 'ad_id' keys,
return rows where the path partition ad_id differs from the internal field.
"""
return [
row for row in rows
if extract_path_ad_id(row["file_path"]) not in (None, row["ad_id"])
]
# ── T014: Field renaming ──────────────────────────────────────────────────────
def rename_columns(df):
"""Rename Parquet camelCase fields to snake_case for Aurora target schema."""
return df.withColumnRenamed("processedAt", "processed_at")
# ── T017: Secrets Manager password retrieval ──────────────────────────────────
def get_db_credentials(secret_arn: str, region: str) -> dict:
client = boto3.client("secretsmanager", region_name=region)
response = client.get_secret_value(SecretId=secret_arn)
return json.loads(response["SecretString"])
# ── T018: DDL (used as JDBC preActions) ──────────────────────────────────────
_CREATE_TABLE_DDL = """
CREATE TABLE IF NOT EXISTS ad_clicks (
ad_id TEXT,
click_id TEXT,
impression_id TEXT,
user_id TEXT,
timestamp BIGINT,
timestamp_iso TEXT,
source_url TEXT,
processed_at TEXT,
sqs_message_id TEXT,
year SMALLINT,
month SMALLINT,
day SMALLINT
)
""".strip()
_PRE_ACTIONS = f"{_CREATE_TABLE_DDL}; TRUNCATE TABLE ad_clicks"
# ── Main job entry point ──────────────────────────────────────────────────────
def main():
from pyspark.sql import SparkSession
from pyspark.sql.functions import input_file_name, udf, col
from pyspark.sql.types import IntegerType
s3_source = os.environ["S3_SOURCE_PATH"]
aurora_endpoint = os.environ["AURORA_ENDPOINT"]
db_name = os.environ.get("DB_NAME", "adclicks")
aws_region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
# T016: SparkSession init
spark = SparkSession.builder \
.appName("ad-clicks-parquet-to-aurora") \
.getOrCreate()
spark.sparkContext.setLogLevel("WARN")
start_time = time.time()
logger.info("Reading Parquet files from: %s", s3_source)
# T016: Read all Parquet files recursively, disabling Hive partition inference.
# The path has mixed structure (plain year/month/day + Hive-style ad_id=xxx)
# which causes Spark's automatic partition discovery to fail with
# "Conflicting directory structures". We use our own UDFs instead.
df = spark.read \
.option("recursiveFileLookup", "true") \
.option("mergeSchema", "true") \
.parquet(s3_source)
# Add file path column for partition extraction and ad_id validation
df = df.withColumn("_file_path", input_file_name())
# T013: Register path parsing UDFs
year_udf = udf(extract_year, IntegerType())
month_udf = udf(extract_month, IntegerType())
day_udf = udf(extract_day, IntegerType())
path_ad_id_udf = udf(extract_path_ad_id)
# Add partition columns from path
df = df \
.withColumn("year", year_udf(col("_file_path")).cast("smallint")) \
.withColumn("month", month_udf(col("_file_path")).cast("smallint")) \
.withColumn("day", day_udf(col("_file_path")).cast("smallint"))
# T015: Detect and warn about ad_id path vs internal field mismatches
df = df.withColumn("_path_ad_id", path_ad_id_udf(col("_file_path")))
mismatches_df = df.filter(
col("_path_ad_id").isNotNull() & (col("_path_ad_id") != col("ad_id"))
)
mismatch_count = mismatches_df.count()
if mismatch_count > 0:
logger.warning("WARN: %d rows have ad_id path/field mismatch — using internal field value", mismatch_count)
mismatches_df.select("_file_path", "_path_ad_id", "ad_id").show(20, truncate=False)
# Count S3 source rows BEFORE dropping helper columns
s3_row_count = df.count()
logger.info("S3 total rows: %d", s3_row_count)
# T014: Rename camelCase → snake_case, drop helper columns
df = rename_columns(df).drop("_file_path", "_path_ad_id")
# T017: Read DB credentials from env vars (pre-fetched by submit_job.py).
# This avoids requiring a Secrets Manager VPC endpoint in the private subnet.
db_user = os.environ["DB_USER"]
db_password = os.environ["DB_PASSWORD"]
jdbc_url = f"jdbc:postgresql://{aurora_endpoint}:5432/{db_name}"
# T018: Write to Aurora with idempotent preActions (CREATE + TRUNCATE)
logger.info("Writing to Aurora: %s/%s", aurora_endpoint, db_name)
df.write \
.format("jdbc") \
.option("url", jdbc_url) \
.option("dbtable", "ad_clicks") \
.option("user", db_user) \
.option("password", db_password) \
.option("driver", "org.postgresql.Driver") \
.option("preActions", _PRE_ACTIONS) \
.option("batchsize", 10000) \
.mode("append") \
.save()
# T019: Post-migration row count validation
aurora_count_df = spark.read \
.format("jdbc") \
.option("url", jdbc_url) \
.option("dbtable", "(SELECT COUNT(*) AS cnt FROM ad_clicks) t") \
.option("user", db_user) \
.option("password", db_password) \
.option("driver", "org.postgresql.Driver") \
.load()
aurora_row_count = aurora_count_df.collect()[0]["cnt"]
duration = round(time.time() - start_time, 1)
print(
f"S3 rows: {s3_row_count} | "
f"Aurora rows: {aurora_row_count} | "
f"Warnings: {mismatch_count} | "
f"Duration: {duration}s"
)
if s3_row_count != aurora_row_count:
raise RuntimeError(
f"Row count mismatch: S3={s3_row_count}, Aurora={aurora_row_count}"
)
logger.info("Migration completed successfully")
spark.stop()
if __name__ == "__main__":
main()
"""
Submit PySpark migration job to EMR Serverless and wait for completion.
Usage: python3 scripts/submit_job.py --stack-name NAME --region REGION --db-name DB
"""
import argparse
import json
import os
import subprocess
import sys
import time
import urllib.request
import boto3
JDBC_JAR_VERSION = "42.7.3"
JDBC_JAR_NAME = f"postgresql-{JDBC_JAR_VERSION}.jar"
JDBC_JAR_URL = (
f"https://repo1.maven.org/maven2/org/postgresql/postgresql/"
f"{JDBC_JAR_VERSION}/{JDBC_JAR_NAME}"
)
S3_SOURCE = "s3://ad-clicks-raw-dev-262969866776/parquet-clicks/"
POLL_INTERVAL = 20
def cfn_output(client, stack_name, key):
resp = client.describe_stacks(StackName=stack_name)
for o in resp["Stacks"][0]["Outputs"]:
if o["OutputKey"] == key:
return o["OutputValue"]
raise KeyError(f"CloudFormation output '{key}' not found in stack '{stack_name}'")
def s3_upload(s3_client, local_path, bucket, key):
print(f" upload: {local_path} → s3://{bucket}/{key}")
s3_client.upload_file(local_path, bucket, key)
def download_jar(path):
if os.path.exists(path):
print(f" JDBC JAR cached: {path}")
return
print(f" Downloading {JDBC_JAR_URL} ...")
urllib.request.urlretrieve(JDBC_JAR_URL, path)
print(f" Saved: {path}")
def wait_for_job(emr_client, app_id, job_run_id):
terminal = {"SUCCESS", "FAILED", "CANCELLED"}
while True:
resp = emr_client.get_job_run(applicationId=app_id, jobRunId=job_run_id)
state = resp["jobRun"]["state"]
print(f" [{time.strftime('%H:%M:%S')}] status: {state}")
if state in terminal:
return state, resp["jobRun"].get("stateDetails", "")
time.sleep(POLL_INTERVAL)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--stack-name", default="ad-clicks-migration")
parser.add_argument("--region", default="us-east-1")
parser.add_argument("--db-name", default="adclicks")
args = parser.parse_args()
region = args.region
stack_name = args.stack_name
db_name = args.db_name
cfn = boto3.client("cloudformation", region_name=region)
s3 = boto3.client("s3", region_name=region)
emr = boto3.client("emr-serverless", region_name=region)
print("==> Reading CloudFormation outputs...")
bucket = cfn_output(cfn, stack_name, "ArtifactsBucket")
app_id = cfn_output(cfn, stack_name, "EmrApplicationId")
role_arn = cfn_output(cfn, stack_name, "EmrJobRoleArn")
aurora_ep = cfn_output(cfn, stack_name, "AuroraEndpoint")
secret_arn = cfn_output(cfn, stack_name, "SecretArn")
print(f" Artifacts bucket : {bucket}")
print(f" EMR app ID : {app_id}")
print(f" Aurora endpoint : {aurora_ep}")
print("\n==> Uploading PySpark script...")
s3_upload(s3, "scripts/migrate.py", bucket, "scripts/migrate.py")
print("\n==> Uploading PostgreSQL JDBC JAR...")
download_jar(JDBC_JAR_NAME)
s3_upload(s3, JDBC_JAR_NAME, bucket, f"jars/{JDBC_JAR_NAME}")
# Retrieve DB credentials locally (submit_job.py runs with local IAM creds).
# Passing them as env vars avoids needing a Secrets Manager VPC endpoint
# inside the private EMR subnet.
print(" Fetching DB credentials from Secrets Manager...")
sm = boto3.client("secretsmanager", region_name=region)
secret = json.loads(
sm.get_secret_value(SecretId=secret_arn)["SecretString"]
)
db_user = secret["username"]
db_password = secret["password"]
env_vars = {
"S3_SOURCE_PATH": S3_SOURCE,
"AURORA_ENDPOINT": aurora_ep,
"DB_NAME": db_name,
"AWS_DEFAULT_REGION": region,
"DB_USER": db_user,
"DB_PASSWORD": db_password,
}
conf_entries = []
for k, v in env_vars.items():
conf_entries.append(f"--conf spark.emr-serverless.driverEnv.{k}={v}")
conf_entries.append(f"--conf spark.emr-serverless.executorEnv.{k}={v}")
spark_params = " ".join([
f"--jars s3://{bucket}/jars/{JDBC_JAR_NAME}",
*conf_entries,
"--conf spark.driver.memory=2g",
"--conf spark.executor.memory=2g",
"--conf spark.executor.cores=2",
])
job_driver = {
"sparkSubmit": {
"entryPoint": f"s3://{bucket}/scripts/migrate.py",
"sparkSubmitParameters": spark_params,
}
}
monitoring = {
"s3MonitoringConfiguration": {
"logUri": f"s3://{bucket}/logs/"
}
}
print("\n==> Submitting EMR Serverless job...")
resp = emr.start_job_run(
applicationId=app_id,
executionRoleArn=role_arn,
jobDriver=job_driver,
configurationOverrides={"monitoringConfiguration": monitoring},
name="ad-clicks-parquet-to-aurora",
)
job_run_id = resp["jobRunId"]
print(f" Job ID: {job_run_id}")
print("\n==> Waiting for job to complete...")
final_state, details = wait_for_job(emr, app_id, job_run_id)
print(f"\n Final status: {final_state}")
if final_state == "SUCCESS":
print("==> Migration job completed successfully")
print(f"\nLogs: s3://{bucket}/logs/applications/{app_id}/jobs/{job_run_id}/")
sys.exit(0)
else:
print(f"==> Migration job FAILED: {details}", file=sys.stderr)
print(f"Logs: s3://{bucket}/logs/applications/{app_id}/jobs/{job_run_id}/", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
完成 migrate 之後，就可以從 PostgreSQL 撈到跟 S3 一樣的資料。
Medium: 使用 Serverless 架構設計廣告點擊系統 — 用 EMR 將資料一次性遷移到 Aurora PostgreSQL 的完整紀錄
iThome鐵人賽
看影片追技術