Amazon EMR Serverless를 사용하여 Spark 작업 실행

2024년 처음 100개의 스테이션 파일을 다운로드합니다.

$ mkdir -p ~/gsod && cd ~/gsod
aws s3 ls s3://noaa-gsod-pds/2024/ --no-sign-request | head -100 | awk '{print $NF}' > files.txt
while read f; do
    aws s3 cp "s3://noaa-gsod-pds/2024/$f" "$f" --no-sign-request --only-show-errors
done < files.txt
ls | wc -l

이 명령은 총 약 7~8MB에 해당하는 약 100개의 CSV 파일을 다운로드합니다.

gsod/2024/ 접두사 아래의 액세스 포인트에 파일을 업로드합니다. access-point-alias를 액세스 포인트 별칭으로 바꿉니다.

$ aws s3 cp ~/gsod/ "s3://access-point-alias/gsod/2024/" --recursive --exclude "files.txt" --only-show-errors

다음 스크립트를 라는 파일에 저장합니다gsod_monthly.py.

# gsod_monthly.py
import sys
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

INPUT_PATH, OUTPUT_PATH = sys.argv[1], sys.argv[2]

# GSOD sentinels for missing data
TEMP_SENTINEL = 9999.9
PRCP_SENTINEL = 99.99

spark = SparkSession.builder.appName("gsod-monthly-summary").getOrCreate()

raw = spark.read.option("header", True).csv(INPUT_PATH)

cleaned = (raw
    .select(
        F.col("STATION").alias("station"),
        F.col("NAME").alias("station_name"),
        F.col("LATITUDE").cast("double").alias("lat"),
        F.col("LONGITUDE").cast("double").alias("lon"),
        F.to_date("DATE", "yyyy-MM-dd").alias("date"),
        F.col("TEMP").cast("double").alias("temp_f"),
        F.col("PRCP").cast("double").alias("prcp_in"),
        F.col("FRSHTT").alias("frshtt"),
    )
    .filter(F.col("temp_f") != TEMP_SENTINEL)
    .withColumn("month", F.date_format("date", "yyyy-MM"))
    .withColumn(
        "prcp_in",
        F.when(F.col("prcp_in") == PRCP_SENTINEL, None).otherwise(F.col("prcp_in")),
    )
    # FRSHTT is a 6-char bitfield: Fog, Rain, Snow, Hail, Thunder, Tornado.
    # Check only positions 2-4 (Rain, Snow, Hail) for precipitation events.
    .withColumn(
        "had_precip_event",
        F.when(F.col("frshtt").substr(2, 3).rlike("1"), 1).otherwise(0),
    )
)

monthly = (cleaned
    .groupBy("station", "station_name", "lat", "lon", "month")
    .agg(
        F.avg("temp_f").alias("avg_temp_f"),
        F.min("temp_f").alias("min_temp_f"),
        F.max("temp_f").alias("max_temp_f"),
        F.sum("prcp_in").alias("total_prcp_in"),
        F.sum("had_precip_event").alias("precip_event_days"),
        F.count("*").alias("observation_days"),
    )
)

(monthly.write
    .mode("overwrite")
    .partitionBy("month")
    .parquet(OUTPUT_PATH))

spark.stop()

scripts/ 접두사 아래의 액세스 포인트에 스크립트를 업로드합니다.

$ aws s3 cp gsod_monthly.py "s3://access-point-alias/scripts/gsod_monthly.py"

최신 Amazon EMR 릴리스를 사용하여 Spark 애플리케이션을 생성합니다.

$ aws emr-serverless create-application \
    --name fsxn-emr-app --type SPARK --release-label emr-7.0.0

응답의 applicationId에 주의하세요.

애플리케이션을 시작합니다. 시작하면 콜드 스타트 지연 없이 첫 번째 작업이 실행되도록 소규모 작업자 풀이 사전 워밍됩니다.

$ aws emr-serverless start-application --application-id application-id

상태가가 될 때까지 기다립니다STARTED.

$ aws emr-serverless get-application --application-id application-id \
    --query 'application.state'

작업 드라이버 구성을 로 저장합니다job-driver.json. 자리 표시자를 바꿉니다.

{
    "sparkSubmit": {
        "entryPoint": "s3://access-point-alias/scripts/gsod_monthly.py",
        "entryPointArguments": [
            "s3://access-point-alias/gsod/2024/",
            "s3://access-point-alias/gsod-monthly/"
        ],
        "sparkSubmitParameters": "--conf spark.executor.cores=2 --conf spark.executor.memory=4g --conf spark.driver.cores=2 --conf spark.driver.memory=4g --conf spark.executor.instances=2"
    }
}

다음 모니터링 구성을 로 저장합니다job-config.json. 드라이버 및 실행기 로그를 CloudWatch Logs로 전송합니다.

{
    "monitoringConfiguration": {
        "cloudWatchLoggingConfiguration": {
            "enabled": true,
            "logGroupName": "/aws/emr-serverless/fsxn-emr-app"
        }
    }
}

작업을 제출합니다.

$ aws emr-serverless start-job-run \
    --application-id application-id \
    --execution-role-arn arn:aws:iam::account-id:role/fsxn-emr-job-role \
    --name gsod-monthly \
    --job-driver file://job-driver.json \
    --configuration-overrides file://job-config.json

응답의 jobRunId에 주의하세요.

작업 상태를 폴링합니다. 작업은에서 SCHEDULEDRUNNING로 전환됩니다SUCCESS.

$ aws emr-serverless get-job-run \
    --application-id application-id \
    --job-run-id job-run-id \
    --query 'jobRun.state'

출력 파티션을 나열합니다.

$ aws s3 ls "s3://access-point-alias/gsod-monthly/" --recursive

month=YYYY-MM/ 파티션당 하나의 Parquet 파일과 루트에 _SUCCESS 마커가 표시되어야 합니다.

로컬에서 파티션을 읽고 콘텐츠를 확인합니다.

$ aws s3 cp "s3://access-point-alias/gsod-monthly/month=2024-06/" . \
    --recursive --exclude "_SUCCESS"
python3 -c "import pyarrow.parquet as pq; \
    t = pq.read_table(next(__import__('glob').iglob('*.parquet'))); \
    print(t.schema); print(t.to_pandas().head())"

출력 스키마에는 station, , station_name, lat, lon, avg_temp_f, max_temp_f, min_temp_ftotal_prcp_in, 및 precip_event_days가 포함됩니다observation_days.