Deklarative Spark-Pipelines

Spark Declarative Pipelines (SDP) ist ein deklaratives Framework für die Erstellung von Batch- und Streaming-Datenpipelines in 6.0. AWS Glue Mit SDP definieren Sie mithilfe von SQL oder Python, wie Ihre Daten aussehen sollen. Das Framework bestimmt automatisch den Ausführungsplan, löst Abhängigkeiten zwischen Datensätzen auf und führt unabhängige Zweige parallel aus.

SDP vereinfacht die Pipeline-Entwicklung, indem es unumgänglichen Standardcode für Lesen, Schreiben, Katalogregistrierung und Ausführungsreihenfolge überflüssig macht. Sie konzentrieren sich auf Geschäftstransformationen, während das Framework die Pipeline-Infrastruktur verwaltet.

SDP ist in AWS Glue Version 6.0 und höher verfügbar.

SDP-Konzepte

Eine Pipeline besteht aus einer YAML-Manifestdatei (spark-pipeline.yml) und einer oder mehreren SQL- oder Python-Transformationsdateien. SDP automatisch:

Typen von Datensätzen

Drei Datensatztypen sind in SDP verfügbar:

Voraussetzungen

Um SDP verwenden zu können, benötigen Sie Folgendes:

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

Eine Pipeline erstellen

Führen Sie die folgenden Schritte aus, um eine SDP-Pipeline zu erstellen.

Schritt 1: Erstellen Sie die Pipeline YAML

Erstellen Sie eine Datei namens 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"

In der folgenden Tabelle werden die YAML-Felder der Pipeline beschrieben.

Schritt 2: Transformationen schreiben

Erstellen Sie Transformationsdateien in einem transformations/ Verzeichnis. Sie können SQL, Python oder beide in derselben Pipeline verwenden.

SQL-Beispiel (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;

Python-Beispiel (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/")

Beispiel für eine Python-Streaming-Tabelle (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/")
    )

Schritt 3: Auf Amazon S3 hochladen

Laden Sie Ihre Pipeline-Dateien wie folgt auf Amazon S3 hoch:

Schritt 4: Erstellen und starten Sie AWS Glue Auftrag

Erstellen Sie einen AWS Glue Job mit den folgenden Parametern:

Das folgende Beispiel erstellt einen SDP-Job mithilfe der AWS CLI:

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"
  }'

Pipelines ausführen

Sie führen SDP-Pipelines mit aus. StartJobRun Sie können das Ausführungsverhalten mit Jobargumenten steuern, die zur Laufzeit übergeben werden.

Modi ausführen

Übergeben Sie die folgenden Argumente an, StartJobRun um die Pipeline-Ausführung zu steuern:

Verwenden von Iceberg-Tabellen mit SDP

Verwenden Sie Apache Iceberg für Streaming-Tabellen, die eine inkrementelle Cross-Run-Verarbeitung erfordern. Hive-managed Streaming-Tabellen speichern Metadaten lokal und bleiben nicht über Jobausführungen hinweg erhalten.

Um Iceberg zu konfigurieren, fügen Sie Ihrem spark-pipeline.yml Konfigurationsabschnitt Folgendes hinzu:

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"

Wenn Iceberg konfiguriert ist, erhalten Sie die folgenden Vorteile:

Überlegungen und Einschränkungen

Beachten Sie Folgendes, wenn Sie SDP verwenden:

Migration von imperativen Skripten zu SDP

Sie können bestehende imperative Spark-Skripte schrittweise zu SDP migrieren:

SDP kann auf jede Tabelle verweisen, auf die über die zugegriffen werden kann SparkSession, einschließlich vorhandener Datenkatalogtabellen, externer Tabellen und datenbankübergreifender Verweise.