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

Apache Iceberg ist das empfohlene Tabellenformat für Streaming-Tabellen, die eine dauerhafte, laufübergreifende inkrementelle Verarbeitung erfordern, da es nicht vom dateibasierten _spark_metadata Protokoll abhängt, das von Hive oder von -verwalteten Tabellen verwendet wird. AWS Glue Sie können Iceberg mit SDP auf zwei Arten konfigurieren, je nachdem, ob Ihre Ausgabetabellen im Datenkatalog registriert werden sollen.

Option 1: Iceberg mit dem Datenkatalog (empfohlen)

Verwenden Sie diese Option, wenn Sie möchten, dass Ihre Iceberg-Tabellen im Datenkatalog (mittable_type=ICEBERG) registriert werden, sodass sie von anderen Engines wie Amazon Redshift und Amazon EMR aus abgefragt werden können. Tabellendaten und Metadaten werden in Amazon S3 gespeichert, und der inkrementelle Cross-Run-Status wird beibehalten. Fügen Sie dem configuration Abschnitt Ihres Folgendes hinzu: spark-pipeline.yml

configuration:
  spark.sql.catalog.glue_catalog: "org.apache.iceberg.spark.SparkCatalog"
  spark.sql.catalog.glue_catalog.catalog-impl: "org.apache.iceberg.aws.glue.GlueCatalog"
  spark.sql.catalog.glue_catalog.io-impl: "org.apache.iceberg.aws.s3.S3FileIO"
  spark.sql.catalog.glue_catalog.warehouse: "s3://my-bucket/iceberg-warehouse"

Legen Sie in Ihrer Pipeline-YAML eine AWS Glue Datenbank fest catalog: glue_catalog und legen database: Sie sie fest. Wenn Sie den AWS Glue Job erstellen, setzen Sie --enable-spark-declarative-pipeline auftrue. Stellen Sie diese Option nicht --enable-glue-datacatalog für Iceberg-Tabellen ein.

Option 2: Iceberg mit einem dateibasierten (Hadoop) Katalog

Verwenden Sie diese Option, wenn Sie keine Datenkatalogregistrierung benötigen. Die Iceberg-Metadaten sind in Amazon S3 dateibasiert, und die Tabellen sind nicht im Datenkatalog registriert. Fügen Sie dem configuration Abschnitt Ihres Folgendes hinzu: 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"

Verwenden Sie Option 1, wenn Sie möchten, dass Ihre Tabellen im Datenkatalog registriert sind und von anderen Engines abgefragt werden können. Verwenden Sie Option 2 nur, wenn Sie keine Datenkatalogregistrierung benötigen.

Wenn Iceberg konfiguriert ist, erhalten Sie die folgenden Vorteile:

Sie können auch inkrementell aus einer Iceberg-Tabelle als Streaming-Quelle lesen. In einer Medallion-Architektur kann eine Downstream-Streaming-Tabelle nur die neuen Zeilen verarbeiten, die bei jedem Lauf in eine Upstream-Iceberg-Tabelle übernommen werden. Das folgende Beispiel liest inkrementell aus einer bronze Iceberg-Tabelle in eine Streaming-Tabelle ein: silver

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

spark = SparkSession.active()

dp.create_streaming_table(
    "silver",
    comment="Incremental silver layer built from the bronze Iceberg table"
)

@dp.append_flow(target="silver")
def from_bronze():
    # Incremental read from the Iceberg bronze table; each run processes only new rows.
    return spark.readStream.table("bronze")

Ü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 den zugegriffen werden kann SparkSession, einschließlich vorhandener Datenkatalogtabellen, externer Tabellen und datenbankübergreifender Verweise.