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 inkrementelle Cross-Run-Verarbeitung mit Streaming-Tabellen: Verwenden Sie Iceberg-Tabellen (Hive-managed Streaming-Tabellen unterstützen keine inkrementelle Cross-Run-Verarbeitung) **Wichtig** Wenn Sie das `database:` Feld in Ihrer Pipeline-YAML verwenden, muss die entsprechende AWS Glue Datenbank auf einen Amazon S3-Pfad `LocationUri` festgelegt sein. Datenbanken, die über die Konsole erstellt wurden, haben oft ein leeres Feld`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 Zieldatenbank für Ausgabetabellen. Die Datenbank muss existieren und einen LocationUri Satz haben. | | 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=...)`. Der `@dp.streaming_table` Dekorateur ist nicht verfügbar. ### 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: `--refresh` Führt alle Datensätze aus. Materialisierte Ansichten werden vollständig neu berechnet. Streaming-Tabellen verarbeiten nur neue Daten seit dem letzten Checkpoint. `--refresh ` Führt nur den angegebenen Datensatz aus. Bei Streaming-Tabellen werden neue Daten inkrementell verarbeitet. Bei materialisierten Ansichten erfolgt dadurch nur eine vollständige Neuberechnung dieser Ansicht. `--full-refresh` Setzt alle Datensätze zurück und berechnet sie neu. Bei Streaming-Tabellen werden dadurch Checkpoints zurückgesetzt und alle Daten werden von Grund auf neu verarbeitet. `--full-refresh-all` Löscht alle Tabellen und verarbeitet die gesamte Pipeline von Grund auf neu. ## 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: + 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 ## Ü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 Sie `dp.create_streaming_table()` mit. `@dp.append_flow(target=...)` Der `@dp.streaming_table` Decorator ist in der aktuellen Version nicht verfügbar. + **Cross-run Für die inkrementelle Verarbeitung ist Iceberg erforderlich ** — Streaming-Tabellen mit Hive oder AWS Glue verwaltetem Katalog unterstützen keine inkrementelle Verarbeitung über mehrere Auftragsläufe hinweg. Verwenden Sie Iceberg-Tabellen für einen dauerhaften inkrementellen Status. + **Datenbank LocationUri erforderlich ** — Wenn Sie `database:` in Ihrer Pipeline eine YAML-Datei angeben, muss die AWS Glue Datenbank auf einen Amazon `LocationUri` S3-Pfad eingestellt sein. Ohne sie schlägt die Pipeline fehl. + **Erwartungen an die Datenqualität ** — Inline-Anmerkungen zur Datenqualität werden im aktuellen SDP-Framework nicht unterstützt. + **`withColumn`Bei Downstream-Abfragefunktionen vermeiden ** — 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 — konvertieren Sie einen einzelnen `spark.sql(...).write.saveAsTable(...)` Aufruf in eine `CREATE MATERIALIZED VIEW` SQL-Anweisung. 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 der Umstellung parallel aus — SDP-Jobs und zwingende Jobs können nebeneinander existieren. SDP kann auf jede Tabelle verweisen, auf die über die zugegriffen werden kann SparkSession, einschließlich vorhandener Datenkatalogtabellen, externer Tabellen und datenbankübergreifender Verweise.