Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Crea pipeline ETL utilizzando AWS Glue

I team di progettazione dei dati spesso utilizzano dati grezzi che arrivano su un volume FSx for ONTAP da applicazioni, perdite giornaliere di file o integrazioni di partner tramite NFS o SMB. La preparazione di tali dati per l'analisi a valle richiede la lettura, la trasformazione, l'arricchimento o il ripartizionamento su larga scala e la messa a disposizione di analisti e applicazioni dell'output curato.

Con un access point Amazon S3 collegato al volume FSx for ONTAP, AWS Glue legge i dati di origine, li trasforma con il tipo di runtime che preferisci (Apache Spark, Python shell o Ray) e riscrive l'output curato sullo stesso volume. Sia i set di dati grezzi che quelli curati rimangono su FSx for ONTAP, quindi le policy di snapshot, backup e conservazione del volume si applicano in modo uniforme a tutta la pipeline. Poiché un volume FSx for ONTAP è accessibile simultaneamente tramite NFS, SMB e l'API Amazon S3, i dati grezzi possono essere prodotti da client NFS o SMB e l'output curato può essere utilizzato da uno qualsiasi di questi protocolli.

In questo tutorial, utilizzi il set di dati di NYC Taxi Trip contenuto nel tutorial. Interroga i file con SQL utilizzando Amazon Athena Un job AWS Glue ETL legge i dati grezzi di Parquet, aggiunge colonne calcolate, filtra i record non validi e riscrive l'output trasformato nel volume partizionato per ora del giorno.

Prerequisiti

Prima di iniziare, assicurati di disporre di:

Fase 1: Creare lo script ETL

PySpark Lo script seguente legge i dati grezzi del viaggio in taxi dal volume FSx for ONTAP, applica le trasformazioni e riscrive i risultati nel volume. Salva questo script come. 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()

Lo script esegue le seguenti trasformazioni:

Passaggio 2: carica lo script e crea il lavoro

Carica lo script ETL sul tuo volume FSx for ONTAP tramite il punto di accesso e crea AWS Glue un job che vi faccia riferimento. AWS Glue carica lo script dal punto di accesso all'avvio del processo, allo stesso modo in cui carica gli script da un bucket Amazon S3 standard.

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

Fase 3: Esegui il processo

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

Monitora lo stato del lavoro. Il lavoro viene in genere completato in uno o due minuti con due G.1X lavoratori.

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

Al termine del processo, verificate l'output trasformato sul volume 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/

L'output è suddiviso in quattro directory in base all'ora del giorno. Ogni partizione contiene file Parquet con i dati trasformati.

Fase 4: Interroga i dati trasformati

Esegui un AWS Glue crawler sull'output trasformato per registrarlo in AWS Glue Data Catalog, quindi interrogalo con 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

Una volta completato il crawler, interroga i dati trasformati in Athena. La struttura partizionata consente ad Athena di scansionare solo le partizioni pertinenti.

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

Poiché i dati sono partizionati datime_of_day, la seconda query analizza solo la morning partizione, riducendo la quantità di dati letti e migliorando le prestazioni delle query.

Considerazioni

Eliminazione

Per evitare addebiti continui, elimina le risorse che hai creato in questo tutorial.

Nell'editor di query Athena, rilasciate la tabella creata dal 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