Pipeline dichiarative Spark

Spark Declarative Pipelines (SDP) è un framework dichiarativo per la creazione di pipeline di dati in batch e streaming nella versione 6.0. AWS Glue Con SDP, definisci come dovrebbero apparire i tuoi dati usando SQL o Python e il framework determina automaticamente il piano di esecuzione, risolve le dipendenze tra i set di dati ed esegue rami indipendenti in parallelo.

SDP semplifica lo sviluppo della pipeline eliminando il codice standard imperativo per la lettura, la scrittura, la registrazione del catalogo e l'ordine di esecuzione. Ti concentri sulle trasformazioni aziendali mentre il framework gestisce l'infrastruttura della pipeline.

SDP è disponibile nella AWS Glue versione 6.0 e successive.

Concetti SDP

Una pipeline è costituita da un file manifest YAML (spark-pipeline.yml) e da uno o più file di trasformazione SQL o Python. SDP automaticamente:

Tipi di set di dati

In SDP sono disponibili tre tipi di set di dati:

Prerequisiti

Per utilizzare SDP, è necessario quanto segue:

aws glue create-database --database-input '{
  "Name":"my_pipeline_db",
  "LocationUri":"s3://my-bucket/warehouse/my_pipeline_db"
}'

Creazione di una pipeline

Per creare una pipeline SDP, completate i seguenti passaggi.

Fase 1: Creare la pipeline YAML

Creare un file denominato spark-pipeline.yml:

name: my_analytics_pipeline
catalog: spark_catalog
database: analytics_db
storage: s3://my-bucket/pipeline-storage/
libraries:
  - glob:
      include: transformations/**
configuration:
  spark.sql.shuffle.partitions: "4"

La tabella seguente descrive i campi YAML della pipeline.

Fase 2: Scrivere le trasformazioni

Crea file di trasformazione in una transformations/ directory. Puoi usare SQL, Python o entrambi nella stessa pipeline.

Esempio SQL ()transformations/silver.sql:

CREATE MATERIALIZED VIEW silver_sales AS
SELECT *, UPPER(region) as clean_region
FROM bronze_sales
WHERE amount > 0;

CREATE MATERIALIZED VIEW gold_summary AS
SELECT clean_region, COUNT(*) as order_count, SUM(amount) as total_revenue
FROM silver_sales
GROUP BY clean_region;

Esempio in Python (transformations/bronze.py):

from pyspark import pipelines as dp
from pyspark.sql import DataFrame, SparkSession

spark = SparkSession.active()

@dp.materialized_view(comment="Raw sales data from S3")
def bronze_sales() -> DataFrame:
    return spark.read.format("csv").option("header", "true") \
        .option("inferSchema", "true") \
        .load("s3://source-bucket/raw-data/sales/")

Esempio di tabella di streaming in Python ()transformations/events.py:

from pyspark import pipelines as dp
from pyspark.sql import DataFrame, SparkSession

spark = SparkSession.active()

dp.create_streaming_table(
    "streaming_events",
    comment="Incremental event ingestion",
    schema="event_id STRING, event_type STRING, timestamp LONG, payload STRING"
)

@dp.append_flow(target="streaming_events")
def ingest_events() -> DataFrame:
    return (
        spark.readStream.format("json")
        .schema("event_id STRING, event_type STRING, timestamp LONG, payload STRING")
        .load("s3://source-bucket/events/")
    )

Fase 3: Caricamento su Amazon S3

Carica i file della pipeline su Amazon S3 in uno dei seguenti modi:

Fase 4: Creare ed eseguire AWS Glue job

Crea un AWS Glue lavoro con i seguenti parametri:

L'esempio seguente crea un job SDP utilizzando la CLI AWS :

aws glue create-job \
  --name my-sdp-pipeline \
  --role arn:aws:iam::123456789012:role/MyGlueRole \
  --glue-version 6.0 \
  --worker-type G.1X --number-of-workers 2 \
  --command '{"Name":"glueetl","ScriptLocation":"s3://my-bucket/pipelines/my_pipeline.zip"}' \
  --default-arguments '{
      "--enable-spark-declarative-pipeline": "true",
      "--enable-glue-datacatalog": "true"
  }'

Esecuzione di pipeline

Si eseguono pipeline SDP utilizzando. StartJobRun È possibile controllare il comportamento di esecuzione con gli argomenti del lavoro passati in fase di esecuzione.

Modalità di esecuzione

Passate i seguenti argomenti a per StartJobRun controllare l'esecuzione della pipeline:

Utilizzo delle tabelle Iceberg con SDP

Per le tabelle di streaming che richiedono un'elaborazione incrementale tra più esecuzioni, usa Apache Iceberg. Hive-managed Le tabelle di streaming archiviano i metadati localmente e non persistono tra le esecuzioni dei processi.

Per configurare Iceberg, aggiungi quanto segue alla tua sezione di configurazione: spark-pipeline.yml

configuration:
  spark.sql.catalog.spark_catalog: "org.apache.iceberg.spark.SparkSessionCatalog"
  spark.sql.catalog.spark_catalog.type: "hadoop"
  spark.sql.catalog.spark_catalog.warehouse: "s3://my-bucket/iceberg-warehouse"

Con Iceberg configurato, ottieni i seguenti vantaggi:

Considerazioni e limitazioni

Quando utilizzate SDP, tenete presente quanto segue:

Migrazione da script imperativi a SDP

Puoi migrare gli script Spark imperativi esistenti su SDP in modo incrementale:

SDP può fare riferimento a qualsiasi tabella accessibile tramite SparkSession, comprese le tabelle esistenti del Data Catalog, le tabelle esterne e i riferimenti tra database.