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. # 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: + Löst Abhängigkeiten auf, indem die DAG aus Tabellenverweisen abgeleitet wird + Bestimmt die Ausführungsreihenfolge ohne manuelle Orchestrierung + Führt unabhängige Zweige parallel aus, um maximalen Durchsatz zu erzielen + Verwaltet den inkrementellen Status für Streaming-Tabellen über Checkpoints + Registriert Ausgabetabellen im Katalog bei der Materialisierung ### Typen von Datensätzen Drei Datensatztypen sind in SDP verfügbar: Streaming-Tabelle Verarbeitet nur neue Daten seit dem letzten Lauf. Behält den Status über alle Jobausführungen hinweg mithilfe von Checkpoints bei. Verwenden Sie Streaming-Tabellen für die Erfassung, Event-Streams, IoT-Daten, die Erfassung von Änderungsdaten und reine Anfügequellen. Materialisierte Ansicht Berechnet den Datensatz bei jedem Lauf vollständig neu. Die Ausgabe gibt immer den aktuellen Status der Quelldaten wieder. Verwenden Sie materialisierte Ansichten für Aggregationen, Verknüpfungen, zusammenfassende Analysen und Berichte. Temporäre Ansicht Session-scoped und nicht gespeichert oder katalogisiert. Verwenden Sie temporäre Ansichten für Zwischentransformationen und Staging-Logik. **Wichtig** Materialisierte Ansichten führen in der aktuellen Version immer eine vollständige Neuberechnung durch. Sie unterstützen keine inkrementelle Aktualisierung. Verwenden Sie Streaming-Tabellen für inkrementelle Workloads. ## Voraussetzungen Um SDP verwenden zu können, benötigen Sie Folgendes: + AWS Glue Version 6.0 + Ein Amazon S3-Standort für die Pipeline-Speicherung (Prüfpunkte, Metadaten) + Für die Datenkatalog-Integration (optional): setzen Sie `--enable-glue-datacatalog` diese Option auf`true`. Alternativ können Sie die Katalogeinstellungen direkt über die Spark-Konfiguration konfigurieren. + Für persistente Tabellenspeicherung: Stellen Sie `spark.sql.warehouse.dir` entweder einen Amazon S3-Pfad ein oder legen Sie das `database:` Feld in der Pipeline-YAML fest und stellen Sie sicher, dass für die AWS Glue Datenbank ein Amazon S3-Pfad `LocationUri` konfiguriert ist + Für die inkrementelle Cross-Run-Verarbeitung mit Streaming-Tabellen müssen die Daten und der Checkpoint-Status einer Streaming-Tabelle auf Amazon S3 bestehen bleiben. Für Hive- oder AWS Glue verwaltete Tabellen (ohne Iceberg) muss die Datenbank beispielsweise auf einen Amazon S3-Pfad eingestellt werden`LocationUri`, während Apache Iceberg-Tabellen ihre Tabellenmetadaten selbst verwalten. **Wichtig** Wenn Sie das `database:` Feld in Ihrer Pipeline-YAML für Hive- oder AWS Glue-verwaltete Tabellen (nicht von Iceberg) verwenden, muss die entsprechende AWS Glue Datenbank auf einen Amazon S3-Pfad eingestellt sein. `LocationUri` Dadurch werden verwaltete Streaming-Tabellen (und ihr `_spark_metadata` Protokoll) auf Amazon S3 platziert, wodurch eine persistente, laufübergreifende inkrementelle Verarbeitung dieser Tabellen ermöglicht wird. `LocationUri` Iceberg-Tabellen, die den Datenkatalog mit `catalog-impl=GlueCatalog` (Option 1) verwenden, benötigen keine Datenbank. `LocationUri` Erstellen oder aktualisieren Sie die Datenbank mit einem expliziten Amazon S3-Speicherort: ``` 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. | Feld | Erforderlich | Beschreibung | | --- | --- | --- | | name | Ja | Ein Name für Ihre Pipeline. | | catalog | Nein | Der zu verwendende Katalog. Standardeinstellung: spark\_catalog. | | database | Nein | Die AWS Glue Datenbank für Ausgabetabellen. Die Anforderungen für LocationUri finden Sie unter [Voraussetzungen](#spark-declarative-pipelines-prerequisites). | | storage | Ja | Ein Amazon S3-Pfad für Pipeline-Checkpoints und Metadaten. | | libraries | Ja | Glob-Muster, die die Transformationsdateien einbeziehen sollen. | | configuration | Nein | Eigenschaften der Spark-Konfiguration. | ### 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/") ) ``` **Anmerkung** Verwenden Sie für Streaming-Tabellen in Python die `dp.create_streaming_table()` Kombination mit`@dp.append_flow(target=...)`. ### Schritt 3: Auf Amazon S3 hochladen Laden Sie Ihre Pipeline-Dateien wie folgt auf Amazon S3 hoch: + Eine `.zip` Datei, die `spark-pipeline.yml` und das `transformations/` Verzeichnis enthält + Ein Amazon S3-Präfix (Verzeichnis), das dieselbe Struktur enthält ### Schritt 4: Erstellen und starten Sie AWS Glue Auftrag Erstellen Sie einen AWS Glue Job mit den folgenden Parametern: + `--enable-spark-declarative-pipeline`: `true` (erforderlich; aktiviert den SDP-Modus) + `ScriptLocation`: Pipeline-Definition-Zip oder ein Amazon S3-Präfix (für die SDP-Pipeline erforderlich) + `--enable-glue-datacatalog`: `true` (optional; registriert Tabellen im Datenkatalog) 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: `--conf spark.glue.sdp.jobMode` Steuert den Ausführungsmodus: + `RUN`(Standard): Führt die Pipeline normal aus. + `VALIDATE`: Führt einen Probelauf durch, der die YAML-Syntax, die Auflösung von Abhängigkeiten und die SQL/Python Kompilierung überprüft, ohne Daten zu schreiben. `--conf spark.glue.sdp.runMode` Steuert, welche Datensätze aktualisiert werden: Standard (kein Run-Mode-Flag) Führt alle Datensätze aus. Materialisierte Ansichten werden vollständig neu berechnet; Streaming-Tabellen verarbeiten nur neue Daten seit dem letzten Checkpoint. `--refresh ` Aktualisiert nur den angegebenen Datensatz. Streaming-Tabellen verarbeiten neue Daten inkrementell; materialisierte Ansichten werden vollständig neu berechnet. `--full-refresh ` Setzt nur den angegebenen Datensatz zurück und berechnet ihn neu. Bei Streaming-Tabellen wird dadurch der Checkpoint zurückgesetzt und alle Daten werden erneut verarbeitet. `--full-refresh-all` Setzt alle Datensätze zurück und berechnet sie neu. ## 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 (mit`table_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` auf`true`. Stellen Sie diese Option nicht `--enable-glue-datacatalog` für Iceberg-Tabellen ein. **Anmerkung** Der Iceberg-Katalog verwendet seinen eigenen `warehouse` Speicherort für Tabellendaten und Metadaten. Wenn Sie diese Option verwenden, müssen Sie keine Datenbank `LocationUri` für die Iceberg-Tabellen selbst einrichten`spark.sql.warehouse.dir`. ### 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" ``` **Anmerkung** Beachten Sie zu dieser Option Folgendes: Tabellen, die mit dieser Option erstellt wurden, sind nicht im Datenkatalog registriert und können daher nicht von Abfrage-Engines wie abgefragt werden. Diese Option verwendet einen eigenen `warehouse` Speicherort für die Tabellenspeicherung. 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. **Anmerkung** Der `--enable-glue-datacatalog` Job-Parameter verbindet den Spark Hive-Metastore mit dem Datenkatalog für Hive-Tabellen (ohne Iceberg). `catalog-impl=GlueCatalog`Registriert Tabellen bei Iceberg-Tabellen über das AWS SDK direkt im Datenkatalog, sodass Sie keine Einstellungen für Iceberg vornehmen. `--enable-glue-datacatalog` Konfigurieren Sie Iceberg nicht mit der Standardeinstellung `SparkSessionCatalog` (`type: hive`) zusammen mit dem `--enable-glue-datacatalog` Versuch, Iceberg-Tabellen im Datenkatalog zu registrieren: In AWS Glue 6.0 schlägt diese Kombination fehl. Wenn Iceberg konfiguriert ist, erhalten Sie die folgenden Vorteile: + Streaming-Tabellen behalten den Checkpoint-Status in Amazon S3 bei allen Auftragsausführungen bei + Bei jedem Lauf werden neue Datendateien und Iceberg-Snapshots erstellt + Nachfolgende Läufe werden ab dem letzten festgeschriebenen Offset fortgesetzt + Der vollständige Tabellenverlauf wird durch den Snapshot-Mechanismus von Iceberg gespeichert 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: + **Materialisierte Ansichten werden immer vollständig neu berechnet. ** Eine inkrementelle Aktualisierung wird nicht unterstützt. Verwenden Sie Streaming-Tabellen für inkrementelle Workloads. + **Python-API für Streaming-Tabellen. ** Verwenden der `dp.create_streaming_table()` mit `@dp.append_flow(target=...)`. + **Cross-run Inkrementelle Verarbeitung für Streaming-Tabellen**. Streaming-Tabellen unterstützen die inkrementelle Cross-Run-Verarbeitung nur, wenn ihre Daten und ihr Checkpoint-Status auf Amazon S3 bestehen bleiben. Hive- oder AWS Glue verwaltete Tabellen (nicht von Iceberg) erfordern, dass die Datenbank auf einen Amazon S3-Pfad eingestellt ist`LocationUri`, während Apache Iceberg-Tabellen ihre Tabellenmetadaten selbst verwalten. + **Für Hive- oder AWS Glue-verwaltete Tabellen (nicht von Iceberg) ist eine Datenbank LocationUri erforderlich. ** Iceberg-Tabellen, mit `catalog-impl=GlueCatalog` denen der Datenkatalog verwendet wird, benötigen keinen. Details hierzu finden Sie unter [Voraussetzungen](#spark-declarative-pipelines-prerequisites). + **Erwartungen ** an die Datenqualität. Inline-Anmerkungen zur Datenqualität werden im aktuellen SDP-Framework nicht unterstützt. + **Vermeiden Sie es `withColumn` in nachgelagerten Abfragefunktionen. ** Wenn ein Downstream-Datensatz (z. B. eine materialisierte Ansicht) mithilfe `spark.table(...)` und Anwendung aus einem Upstream-Pipeline-Datensatz liest und anwendet`.withColumn(...)`, erkennt SDP möglicherweise nicht die Abhängigkeit zwischen den Datensätzen beim zweiten und nachfolgenden Durchlauf. Dies führt dazu, dass der Downstream veraltete Daten aus dem vorherigen Lauf liest (Verzögerung bei einem Durchlauf). Um dieses Problem zu vermeiden, drücken Sie abgeleitete Spalten nach innen aus, `.select(...)` anstatt sie zu verwenden. `.withColumn(...)` Vermeiden Sie außerdem alle Operationen, die eine Planauflösung erzwingen (z. B. `.schema` oder`.collect`) innerhalb von Abfragefunktionen. + **Keine Migrationstools. ** Die automatische Migration von anderen Pipeline-Frameworks wird nicht unterstützt. Migrieren Sie Tabellen inkrementell; SDP kann aus vorhandenen Katalogtabellen lesen. + **Planung. ** SDP-Jobs verwenden dieselben Planungsmechanismen wie andere AWS Glue Jobs (AWS Glue Trigger, Amazon EventBridge, Apache Airflow). ## Migration von imperativen Skripten zu SDP Sie können bestehende imperative Spark-Skripte schrittweise zu SDP migrieren: 1. Beginnen Sie mit einer Tabelle, indem Sie einen einzelnen `spark.sql(...).write.saveAsTable(...)` Aufruf in eine `CREATE MATERIALIZED VIEW` SQL-Anweisung konvertieren. 1. Fügen Sie Tabellen schrittweise hinzu. SDP verarbeitet gemischte Abhängigkeiten. SDP-Tabellen können aus vorhandenen Katalogtabellen lesen, die nicht Teil der Pipeline sind. 1. Führen Sie beide Muster während des Übergangs parallel aus. SDP-Jobs und zwingende Jobs können koexistieren. SDP kann auf jede Tabelle verweisen, auf die über den zugegriffen werden kann SparkSession, einschließlich vorhandener Datenkatalogtabellen, externer Tabellen und datenbankübergreifender Verweise.