As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Crie pipelines de ETL usando AWS Glue

As equipes de engenharia de dados geralmente têm dados brutos chegando em um volume FSx for ONTAP de aplicativos, descartes diários de arquivos ou integrações de parceiros via NFS ou SMB. Preparar esses dados para análises posteriores exige lê-los, transformá-los, enriquecê-los ou reparticioná-los em grande escala e disponibilizar a saída selecionada para analistas e aplicativos.

Com um ponto de acesso Amazon S3 conectado ao volume FSx for ONTAP, AWS Glue lê os dados de origem, os transforma com sua escolha de tempo de execução (Apache Spark, Python shell ou Ray) e grava a saída selecionada de volta no mesmo volume. Tanto os conjuntos de dados brutos quanto os selecionados permanecem no FSx for ONTAP, portanto, as políticas de snapshot, backup e retenção do volume se aplicam uniformemente em todo o pipeline. Como um volume FSx for ONTAP pode ser acessado simultaneamente por NFS, SMB e pela API do Amazon S3, os dados brutos podem ser produzidos por clientes NFS ou SMB e a saída selecionada pode ser consumida por qualquer um desses protocolos.

Neste tutorial, você usa o conjunto de dados de viagem de táxi de Nova York do Consulte arquivos com SQL usando o Amazon Athena tutorial. Uma tarefa de AWS Glue ETL lê os dados brutos do Parquet, adiciona colunas computadas, filtra registros inválidos e grava a saída transformada de volta no volume particionado por hora do dia.

Pré-requisitos

Antes de começar, você deve ter o seguinte:

Etapa 1: criar o script ETL

O PySpark script a seguir lê os dados brutos da viagem de táxi do seu volume FSx for ONTAP, aplica transformações e grava os resultados de volta no volume. Salve esse script comotaxi_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()

O script executa as seguintes transformações:

Etapa 2: Carregar o script e criar o trabalho

Carregue o script ETL em seu volume FSx for ONTAP por meio do ponto de acesso e crie AWS Glue um trabalho que faça referência a ele. AWS Glue carrega o script do ponto de acesso na inicialização do trabalho, da mesma forma que carrega scripts de um bucket padrão do Amazon S3.

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

Etapa 3: Executar o trabalho

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

Monitore o status do trabalho. O trabalho normalmente é concluído em um a dois minutos com dois G.1X trabalhadores.

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

Quando o trabalho for concluído, verifique a saída transformada em seu FSx para o 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/

A saída é particionada em quatro diretórios por hora do dia. Cada partição contém arquivos Parquet com os dados transformados.

Etapa 4: consultar os dados transformados

Execute um AWS Glue rastreador na saída transformada para registrá-la no e, em seguida AWS Glue Data Catalog, consulte-a com o 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

Depois que o rastreador for concluído, consulte os dados transformados no Athena. A estrutura particionada permite que o Athena digitalize somente as partições relevantes.

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

Como os dados são particionados portime_of_day, a segunda consulta verifica somente a morning partição, reduzindo a quantidade de dados lidos e melhorando o desempenho da consulta.

Considerações

Limpeza

Para evitar cobranças contínuas, exclua os recursos que você criou neste tutorial.

No editor de consultas do Athena, solte a tabela criada pelo rastreador:

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