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

Pour les tables de streaming qui nécessitent un traitement incrémentiel croisé, utilisez Apache Iceberg. Hive-managed les tables de streaming stockent les métadonnées localement et ne persistent pas d'une exécution à l'autre.

Pour configurer Iceberg, ajoutez ce qui suit à votre section spark-pipeline.yml de configuration :

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"

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

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.