Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Erstellen Sie ETL-Pipelines mit AWS Glue

Datenentwicklungsteams haben häufig Rohdaten, die aus Anwendungen, täglichen Dateiabzügen oder Partnerintegrationen über NFS oder SMB stammen, auf einem FSx for ONTAP-Volume landen. Die Vorbereitung dieser Daten für nachgelagerte Analysen erfordert das Lesen, Transformieren, Anreichern oder Neupartitionieren der Daten in großem Umfang und die Bereitstellung der kuratierten Ergebnisse für Analysten und Anwendungen.

Mit einem Amazon S3 S3-Zugriffspunkt, der an das FSx for ONTAP-Volume angeschlossen ist, AWS Glue liest die Quelldaten, transformiert sie mit der Laufzeit Ihrer Wahl (Apache Spark, Python-Shell oder Ray) und schreibt die kuratierte Ausgabe zurück auf dasselbe Volume. Sowohl die Rohdaten als auch die kuratierten Datensätze verbleiben auf FSx for ONTAP, sodass die Snapshot-, Backup- und Aufbewahrungsrichtlinien des Volumes einheitlich für die gesamte Pipeline gelten. Da auf ein FSx for ONTAP-Volume gleichzeitig über NFS, SMB und die Amazon S3 S3-API zugegriffen werden kann, können die Rohdaten von NFS- oder SMB-Clients erzeugt werden und die kuratierte Ausgabe kann von jedem dieser Protokolle verarbeitet werden.

In diesem Tutorial verwenden Sie den NYC Taxi Trip-Datensatz aus dem Tutorial. Dateien mit SQL mit Amazon Athena abfragen Ein AWS Glue ETL-Job liest die Parquet-Rohdaten, fügt berechnete Spalten hinzu, filtert ungültige Datensätze und schreibt die transformierte Ausgabe zurück auf das nach Tageszeit partitionierte Volume.

Voraussetzungen

Stellen Sie vor dem Beginn sicher, dass Sie über das Folgende verfügen:

Schritt 1: Erstellen Sie das ETL-Skript

Das folgende PySpark Skript liest die Rohdaten der Taxifahrten von Ihrem FSx for ONTAP-Volume, wendet Transformationen an und schreibt die Ergebnisse zurück auf das Volume. Speichern Sie dieses Skript unter. 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()

Das Skript führt die folgenden Transformationen durch:

Schritt 2: Laden Sie das Skript hoch und erstellen Sie den Job

Laden Sie das ETL-Skript über den Access Point auf Ihr FSx for ONTAP-Volume hoch und erstellen Sie einen AWS Glue Job, der darauf verweist. AWS Glue lädt das Skript beim Jobstart vom Access Point, genauso wie es Skripts aus einem standardmäßigen Amazon S3 S3-Bucket lädt.

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

Schritt 3: Führen Sie den Job aus

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

Überwachen Sie den Auftragsstatus. Der Auftrag wird in der Regel mit zwei G.1X Mitarbeitern in ein bis zwei Minuten abgeschlossen.

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

Wenn der Job abgeschlossen ist, überprüfen Sie die transformierte Ausgabe auf Ihrem FSx for ONTAP-Volume.

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

Die Ausgabe ist nach Tageszeit in vier Verzeichnisse unterteilt. Jede Partition enthält Parquet-Dateien mit den transformierten Daten.

Schritt 4: Fragen Sie die transformierten Daten ab

Führen Sie einen AWS Glue Crawler für die transformierte Ausgabe aus, um sie in der zu registrieren AWS Glue Data Catalog, und fragen Sie sie dann mit Athena ab.

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

Fragen Sie nach Abschluss des Crawlers die transformierten Daten in Athena ab. Die partitionierte Struktur ermöglicht es Athena, nur die relevanten Partitionen zu scannen.

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

Da die Daten nach partitioniert sindtime_of_day, scannt die zweite Abfrage nur die morning Partition, wodurch die Menge der gelesenen Daten reduziert und die Abfrageleistung verbessert wird.

Überlegungen

Bereinigen

Um laufende Gebühren zu vermeiden, löschen Sie die Ressourcen, die Sie in diesem Tutorial erstellt haben.

Löschen Sie im Athena-Abfrage-Editor die vom Crawler erstellte Tabelle:

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