기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

를 사용하여 ETL 파이프라인 구축 AWS Glue

데이터 엔지니어링 팀은 NFS 또는 SMB를 통한 애플리케이션, 일일 파일 삭제 또는 파트너 통합에서 FSx for ONTAP 볼륨에 대한 원시 데이터 랜딩을 수행하는 경우가 많습니다. 다운스트림 분석을 위해 데이터를 준비하려면 대규모로 데이터를 읽고, 변환하고, 보강하거나, 재분배하고, 큐레이션된 출력을 분석가와 애플리케이션에 제공해야 합니다.

FSx for ONTAP 볼륨에 연결된 Amazon S3 액세스 포인트를 사용하여는 소스 데이터를 AWS Glue 읽고 선택한 런타임(Apache Spark, Python 셸 또는 Ray)으로 변환한 다음 큐레이션된 출력을 동일한 볼륨에 다시 씁니다. 원시 데이터 세트와 큐레이션된 데이터 세트는 모두 FSx for ONTAP에 유지되므로 볼륨의 스냅샷, 백업 및 보존 정책이 파이프라인 전체에 균일하게 적용됩니다. FSx for ONTAP 볼륨은 NFS, SMB 및 Amazon S3 API를 통해 동시에 액세스할 수 있으므로 NFS 또는 SMB 클라이언트에서 원시 데이터를 생성하고 이러한 프로토콜에서 큐레이션된 출력을 사용할 수 있습니다.

이 자습서에서는 자Amazon Athena를 사용하여 SQL로 파일 쿼리습서의 NYC 택시 여행 데이터 세트를 사용합니다. AWS Glue ETL 작업은 원시 Parquet 데이터를 읽고, 계산된 열을 추가하고, 잘못된 레코드를 필터링하고, 변환된 출력을 시간대별로 분할된 볼륨에 다시 씁니다.

사전 조건

시작하기 전에 다음 항목이 준비되었는지 확인합니다.

1단계: ETL 스크립트 생성

다음 PySpark 스크립트는 FSx for ONTAP 볼륨에서 원시 택시 이동 데이터를 읽고, 변환을 적용하고, 결과를 볼륨에 다시 씁니다. 이 스크립트를 로 저장합니다taxi_transform.py.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col, hour, dayofweek, when, round as spark_round

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'AP_ALIAS'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

ap_alias = args['AP_ALIAS']

# Read raw taxi data from FSx through the access point
df = spark.read.parquet(f"s3://{ap_alias}/taxi-data/")

# Transform: filter invalid records, add computed columns
transformed = df \
    .filter(col("trip_distance") > 0) \
    .filter(col("total_amount") > 0) \
    .filter(col("passenger_count") > 0) \
    .withColumn("pickup_hour", hour(col("tpep_pickup_datetime"))) \
    .withColumn("pickup_day_of_week", dayofweek(col("tpep_pickup_datetime"))) \
    .withColumn("cost_per_mile",
        spark_round(col("total_amount") / col("trip_distance"), 2)) \
    .withColumn("time_of_day",
        when(hour(col("tpep_pickup_datetime")).between(6, 11), "morning")
        .when(hour(col("tpep_pickup_datetime")).between(12, 16), "afternoon")
        .when(hour(col("tpep_pickup_datetime")).between(17, 21), "evening")
        .otherwise("night")
    ) \
    .select(
        "tpep_pickup_datetime", "tpep_dropoff_datetime",
        "passenger_count", "trip_distance",
        "PULocationID", "DOLocationID",
        "fare_amount", "tip_amount", "total_amount",
        "pickup_hour", "pickup_day_of_week",
        "cost_per_mile", "time_of_day"
    )

# Write transformed data back to FSx, partitioned by time of day
transformed.write \
    .mode("overwrite") \
    .partitionBy("time_of_day") \
    .parquet(f"s3://{ap_alias}/taxi-data-transformed/")

job.commit()

스크립트는 다음과 같은 변환을 수행합니다.

2단계: 스크립트 업로드 및 작업 생성

액세스 포인트를 통해 ETL 스크립트를 FSx for ONTAP 볼륨에 업로드하고 이를 참조하는 AWS Glue 작업을 생성합니다. 표준 Amazon S3 버킷에서 스크립트를 AWS Glue 로드하는 것과 동일한 방식으로 작업 시작 시 액세스 포인트에서 스크립트를 로드합니다.

$ # Upload the script to the access point
aws s3 cp taxi_transform.py \
    s3://my-ap-alias-ext-s3alias/glue-scripts/taxi_transform.py

# Create the Glue job
aws glue create-job \
    --name fsxn-taxi-transform \
    --role my-glue-role-arn \
    --command '{
        "Name": "glueetl",
        "ScriptLocation": "s3://my-ap-alias-ext-s3alias/glue-scripts/taxi_transform.py",
        "PythonVersion": "3"
    }' \
    --default-arguments '{
        "--AP_ALIAS": "my-ap-alias-ext-s3alias",
        "--job-language": "python"
    }' \
    --glue-version "4.0" \
    --number-of-workers 2 \
    --worker-type "G.1X"

3단계: 작업 실행

$ aws glue start-job-run --job-name fsxn-taxi-transform

작업 상태를 모니터링합니다. 작업은 일반적으로 2명의 G.1X 작업자와 함께 1~2분 내에 완료됩니다.

$ aws glue get-job-runs --job-name fsxn-taxi-transform \
    --query "JobRuns[0].{State:JobRunState,Duration:ExecutionTime,Error:ErrorMessage}"

작업이 완료되면 FSx for ONTAP 볼륨에서 변환된 출력을 확인합니다.

$ aws s3 ls s3://my-ap-alias-ext-s3alias/taxi-data-transformed/
                           PRE time_of_day=afternoon/
                           PRE time_of_day=evening/
                           PRE time_of_day=morning/
                           PRE time_of_day=night/

출력은 시간대별로 네 개의 디렉터리로 분할됩니다. 각 파티션에는 변환된 데이터가 포함된 Parquet 파일이 포함되어 있습니다.

4단계: 변환된 데이터 쿼리

변환된 출력에서 AWS Glue 크롤러를 실행하여에 등록한 AWS Glue Data Catalog다음 Athena로 쿼리합니다.

$ # Create a crawler for the transformed data
aws glue create-crawler \
    --name fsxn-taxi-transformed-crawler \
    --role my-glue-role-arn \
    --database-name fsxn_taxi_demo \
    --targets '{"S3Targets": [{"Path": "s3://my-ap-alias-ext-s3alias/taxi-data-transformed/"}]}'

# Run the crawler
aws glue start-crawler --name fsxn-taxi-transformed-crawler

크롤러가 완료되면 Athena에서 변환된 데이터를 쿼리합니다. 분할된 구조를 통해 Athena는 관련 파티션만 스캔할 수 있습니다.

-- Average cost per mile by time of day
SELECT
    time_of_day,
    COUNT(*) AS trip_count,
    ROUND(AVG(cost_per_mile), 2) AS avg_cost_per_mile,
    ROUND(AVG(tip_amount), 2) AS avg_tip
FROM fsxn_taxi_demo.taxi_data_transformed
GROUP BY time_of_day
ORDER BY trip_count DESC

-- Busiest pickup locations during morning rush
SELECT
    PULocationID AS pickup_location,
    COUNT(*) AS trip_count,
    ROUND(AVG(trip_distance), 2) AS avg_distance
FROM fsxn_taxi_demo.taxi_data_transformed
WHERE time_of_day = 'morning'
GROUP BY PULocationID
ORDER BY trip_count DESC
LIMIT 10

데이터는에 의해 분할되므로 time_of_day두 번째 쿼리는 morning 파티션만 스캔하여 데이터 읽기 양을 줄이고 쿼리 성능을 개선합니다.

고려 사항

정리

요금이 계속 부과되지 않도록 하려면이 자습서에서 생성한 리소스를 삭제합니다.

Athena 쿼리 편집기에서 크롤러가 생성한 테이블을 삭제합니다.

DROP TABLE IF EXISTS fsxn_taxi_demo.taxi_data_transformed;

$ # Delete the Glue job and crawler
aws glue delete-job --name fsxn-taxi-transform
aws glue delete-crawler --name fsxn-taxi-transformed-crawler

# Delete the ETL script and transformed data from the access point
aws s3 rm s3://my-ap-alias-ext-s3alias/glue-scripts/taxi_transform.py
aws s3 rm s3://my-ap-alias-ext-s3alias/taxi-data-transformed/ --recursive

# Delete the IAM role
aws iam delete-role-policy \
    --role-name fsxn-tutorial-glue-etl-role \
    --policy-name glue-fsxn-access
aws iam delete-role --role-name fsxn-tutorial-glue-etl-role