Spark Declarative Pipelines

Spark Declarative Pipelines (SDP) es un marco declarativo para crear canalizaciones de datos por lotes y de transmisión en AWS Glue 6.0. Con SDP, usted define el aspecto que deben tener sus datos mediante SQL o Python y el marco determina automáticamente el plan de ejecución, resuelve las dependencias entre los conjuntos de datos y ejecuta ramificaciones independientes en paralelo.

SDP simplifica el desarrollo de canalizaciones al eliminar el código reutilizable imprescindible para la lectura, la escritura, el registro en el catálogo y el orden de ejecución. Usted se centra en las transformaciones empresariales, mientras que el marco se encarga de la infraestructura de la canalización.

SDP solo está disponible en la versión 6.0 de AWS Glue y las versiones posteriores.

Conceptos de SDP

Una canalización consta de un archivo de manifiesto YAML (spark-pipeline.yml) y uno o más archivos de transformación de SQL o Python. SDP hace lo siguiente automáticamente:

Tipos de conjunto de datos

Hay tres tipos de conjuntos de datos disponibles en SDP:

Requisitos previos

Para utilizar SDP, necesita lo siguiente:

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

Creación de una canalización

Para crear una canalización de SDP, siga los pasos que se describen a continuación.

Paso 1: creación del YAML de canalización

Cree un archivo denominado 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"

En la tabla siguiente se describen los campos del YAML de la canalización.

Paso 2: escritura de transformaciones

Cree archivos de transformación en un directorio transformations/. Puede usar SQL, Python o ambos en la misma canalización.

Ejemplo de 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;

Ejemplo de 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/")

Ejemplo de tabla de transmisión en 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/")
    )

Paso 3: cárguelo en Amazon S3

Cargue sus archivos de canalización en Amazon S3 de una de las siguientes maneras:

Paso 4: creación y ejecución del trabajo de AWS Glue

Cree un trabajo de AWS Glue con los siguientes parámetros:

En el siguiente ejemplo se crea un trabajo de SDP mediante la 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"
  }'

Ejecución de canalizaciones

Para ejecutar canalizaciones de SDP, utilice StartJobRun. Puede controlar el comportamiento de ejecución con los argumentos de trabajo que se pasan en tiempo de ejecución.

Modos de ejecución

Pase los siguientes argumentos a StartJobRun para controlar la ejecución de la canalización:

Utilización de las tablas de Iceberg con SDP

Para las tablas de transmisión que requieren un procesamiento incremental entre ejecuciones, utilice Apache Iceberg. Las tablas de transmisión administradas por Hive almacenan los metadatos de forma local y no persisten entre ejecuciones de los trabajos.

Para configurar Iceberg, agregue lo siguiente a la sección de configuración 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"

Con Iceberg configurado, obtendrá las siguientes ventajas:

Consideraciones y limitaciones

Cuando utilice SDP, tenga en cuenta lo siguiente:

Migración de scripts imperativos a SDP

Puede migrar los scripts imperativos de Spark existentes a SDP de forma incremental:

SDP puede hacer referencia a cualquier tabla a la que se pueda acceder a través de SparkSession, incluidas las tablas del Catálogo de datos existentes, las tablas externas y las referencias entre bases de datos.