Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Membangun pipa ETL menggunakan AWS Glue

Tim rekayasa data sering memiliki pendaratan data mentah pada FSx untuk volume ONTAP dari aplikasi, penurunan file harian, atau integrasi mitra melalui NFS atau SMB. Mempersiapkan data tersebut untuk analitik hilir memerlukan pembacaan, transformasi, pengayaan, atau partisi ulang dalam skala besar, dan membuat output yang dikuratori tersedia untuk analis dan aplikasi.

Dengan jalur akses Amazon S3 yang dilampirkan ke FSx untuk volume ONTAP, AWS Glue membaca data sumber, mengubahnya dengan pilihan runtime Anda (Apache Spark, Python shell, atau Ray), dan menulis output yang dikuratori kembali ke volume yang sama. Baik kumpulan data mentah maupun yang dikurasi tetap ada di FSx untuk ONTAP, sehingga kebijakan snapshot, pencadangan, dan retensi volume berlaku secara seragam di seluruh pipeline. Karena FSx untuk volume ONTAP dapat diakses secara bersamaan melalui NFS, SMB, dan Amazon S3 API, data mentah dapat diproduksi oleh klien NFS atau SMB dan output yang dikuratori dapat dikonsumsi oleh salah satu protokol tersebut.

Dalam tutorial ini, Anda menggunakan dataset perjalanan NYC Taxi dari tutorial. Kueri file dengan SQL menggunakan Amazon Athena Pekerjaan AWS Glue ETL membaca data Parket mentah, menambahkan kolom yang dihitung, memfilter catatan yang tidak valid, dan menulis output yang diubah kembali ke volume yang dipartisi berdasarkan waktu dalam sehari.

Prasyarat

Sebelum Anda mulai, pastikan Anda memiliki yang berikut:

Langkah 1: Buat skrip ETL

PySpark Skrip berikut membaca data perjalanan taksi mentah dari FSx Anda untuk volume ONTAP, menerapkan transformasi, dan menulis hasilnya kembali ke volume. Simpan skrip ini sebagaitaxi_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()

Script melakukan transformasi berikut:

Langkah 2: Unggah skrip dan buat pekerjaan

Unggah skrip ETL ke FSx Anda untuk volume ONTAP melalui titik akses, dan buat pekerjaan yang AWS Glue mereferensikannya. AWS Glue memuat skrip dari titik akses saat startup pekerjaan, dengan cara yang sama memuat skrip dari bucket Amazon S3 standar.

$ # 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"

Langkah 3: Jalankan pekerjaan

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

Pantau status pekerjaan. Pekerjaan biasanya selesai dalam satu hingga dua menit dengan dua G.1X pekerja.

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

Saat pekerjaan selesai, verifikasi output yang diubah pada FSx Anda untuk volume 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/

Output dipartisi menjadi empat direktori berdasarkan waktu hari. Setiap partisi berisi file Parket dengan data yang diubah.

Langkah 4: Kueri data yang diubah

Jalankan AWS Glue crawler pada output yang diubah untuk mendaftarkannya di AWS Glue Data Catalog, lalu kueri dengan 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

Setelah crawler selesai, kueri data yang diubah di Athena. Struktur yang dipartisi memungkinkan Athena untuk memindai hanya partisi yang relevan.

-- 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

Karena data dipartisi olehtime_of_day, kueri kedua hanya memindai morning partisi, mengurangi jumlah data yang dibaca dan meningkatkan kinerja kueri.

Pertimbangan-pertimbangan

Bersihkan

Untuk menghindari biaya yang sedang berlangsung, hapus sumber daya yang Anda buat dalam tutorial ini.

Di editor kueri Athena, jatuhkan tabel yang dibuat oleh crawler:

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