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

Apache Iceberg es el formato de tabla recomendado para las tablas de transmisión que requieren un procesamiento incremental duradero entre ejecuciones, ya que no depende del registro _spark_metadata basado en archivos que utilizan Hive o las tablas administradas por AWS Glue. Puede configurar Iceberg con SDP de dos maneras, dependiendo de si desea que las tablas de salida se registren en el Catálogo de datos.

Opción 1: Iceberg con el Catálogo de datos (recomendado)

Utilice esta opción cuando desee que sus tablas de Iceberg se registren en el Catálogo de datos (con table_type=ICEBERG) para que puedan consultarse desde otros motores, como Amazon Redshift y Amazon EMR. Los datos y metadatos de las tablas se almacenan en Amazon S3 y se conserva el estado incremental entre ejecuciones. Agregue lo siguiente a la sección configuration de 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"

En el YAML de su canalización, establezca catalog: glue_catalog, y establezca database: en una base de datos de AWS Glue. Al crear el trabajo de AWS Glue, establezca --enable-spark-declarative-pipeline en true. No establezca --enable-glue-datacatalog en las tablas de Iceberg.

Opción 2: Iceberg con un catálogo basado en archivos (Hadoop)

Utilice esta opción cuando no necesite el registro del Catálogo de datos. Los metadatos de Iceberg se basan en archivos en Amazon S3, y las tablas no están registradas en el Catálogo de datos. Agregue lo siguiente a la sección configuration de 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"

Utilice la opción 1 si necesita que sus tablas estén registradas en el catálogo de datos y puedan consultarse desde otros motores. Utilice la opción 2 únicamente si no necesita el registro en el Catálogo de datos.

Con Iceberg configurado, obtendrá las siguientes ventajas:

También puede leer de forma incremental una tabla de Iceberg como fuente de transmisión. En una arquitectura tipo medallón, una tabla de transmisión posterior solo puede consumir las nuevas filas confirmadas en una tabla de Iceberg anterior en cada ejecución. En el siguiente ejemplo, se lee de forma incremental de una tabla de Iceberg bronze a una tabla de transmisión 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")

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.