Pipelines déclaratifs Spark

Spark Declarative Pipelines (SDP) est un framework déclaratif permettant de créer des pipelines de données par lots et en streaming dans la version 6.0. AWS Glue Avec SDP, vous définissez à quoi doivent ressembler vos données à l'aide de SQL ou Python, et le framework détermine automatiquement le plan d'exécution, résout les dépendances entre les ensembles de données et exécute des branches indépendantes en parallèle.

SDP simplifie le développement du pipeline en éliminant le code standard impératif pour la lecture, l'écriture, l'enregistrement du catalogue et l'ordre d'exécution. Vous vous concentrez sur les transformations de l'entreprise tandis que le framework gère l'infrastructure du pipeline.

SDP est disponible dans la AWS Glue version 6.0 et les versions ultérieures.

Concepts du SDP

Un pipeline comprend un fichier manifeste YAML (spark-pipeline.yml) et un ou plusieurs fichiers de transformation SQL ou Python. SDP automatiquement :

Types de jeux de données

Trois types de jeux de données sont disponibles dans SDP :

Conditions préalables

Pour utiliser SDP, vous avez besoin des éléments suivants :

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

Création d'un pipeline

Pour créer un pipeline SDP, procédez comme suit.

Étape 1 : Création du pipeline YAML

Créez un fichier nommé 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"

Le tableau suivant décrit les champs YAML du pipeline.

Étape 2 : Écrire des transformations

Créez des fichiers de transformation dans un transformations/ répertoire. Vous pouvez utiliser SQL, Python ou les deux dans le même pipeline.

Exemple 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;

Exemple 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/")

Exemple de table de streaming 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/")
    )

Étape 3 : Chargement sur Amazon S3

Téléchargez vos fichiers de pipeline sur Amazon S3 de la manière suivante :

Étape 4 : Créez et exécutez le AWS Glue tâche

Créez une AWS Glue tâche avec les paramètres suivants :

L'exemple suivant crée une tâche SDP à l'aide de l' AWS interface de ligne de commande :

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

Canalisations en fonctionnement

Vous exécutez des pipelines SDP à l'aide StartJobRun de. Vous pouvez contrôler le comportement d'exécution à l'aide des arguments de tâche transmis au moment de l'exécution.

Modes d'exécution

Transmettez les arguments suivants à StartJobRun pour contrôler l'exécution du pipeline :

Utilisation des tables Iceberg avec SDP

Apache Iceberg est le format de tableau recommandé pour les tables de streaming qui nécessitent un traitement incrémentiel durable et croisé, car il ne dépend pas du _spark_metadata journal basé sur des fichiers utilisé par Hive ou les tables gérées par Hive. AWS Glue Vous pouvez configurer Iceberg avec SDP de deux manières, selon que vous souhaitez que vos tables de sortie soient enregistrées ou non dans le catalogue de données.

Option 1 : Iceberg avec le catalogue de données (recommandé)

Utilisez cette option si vous souhaitez que vos tables Iceberg soient enregistrées dans le catalogue de données (avectable_type=ICEBERG) afin qu'elles puissent être interrogées à partir d'autres moteurs tels qu'Amazon Redshift et Amazon EMR. Les données et les métadonnées des tables sont stockées dans Amazon S3 et l'état incrémentiel des exécutions croisées est préservé. Ajoutez ce qui suit à la configuration section de votre 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"

Dans votre pipeline YAML, définissez catalog: glue_catalog et configurez une AWS Glue base database: de données. Lorsque vous créez la AWS Glue tâche, définissez --enable-spark-declarative-pipeline surtrue. Ne pas installer sur --enable-glue-datacatalog les tables Iceberg.

Option 2 : Iceberg avec un catalogue basé sur des fichiers (Hadoop)

Utilisez cette option lorsque vous n'avez pas besoin d'être enregistré dans le catalogue de données. Les métadonnées Iceberg sont basées sur des fichiers dans Amazon S3 et les tables ne sont pas enregistrées dans le catalogue de données. Ajoutez ce qui suit à la configuration section de votre 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"

Utilisez l'option 1 si vous souhaitez que vos tables soient enregistrées dans le catalogue de données et consultables à partir d'autres moteurs. Utilisez l'option 2 uniquement si vous n'avez pas besoin d'être enregistré dans le catalogue de données.

Une fois Iceberg configuré, vous bénéficiez des avantages suivants :

Vous pouvez également lire de manière incrémentielle à partir d'un tableau Iceberg en tant que source de streaming. Dans une architecture en médaillon, une table de streaming en aval ne peut consommer que les nouvelles lignes qui sont validées dans une table Iceberg en amont à chaque exécution. L'exemple suivant lit de manière incrémentielle une table Iceberg vers une bronze table de silver streaming :

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")

Considérations et restrictions

Lorsque vous utilisez SDP, tenez compte des points suivants :

Migration des scripts impératifs vers SDP

Vous pouvez migrer les scripts Spark impératifs existants vers SDP de manière incrémentielle :

SDP peut référencer n'importe quelle table accessible via le SparkSession, y compris les tables du catalogue de données existantes, les tables externes et les références entre bases de données.