Spark Declarative Pipelines

O Spark Declarative Pipelines (SDP) é uma estrutura declarativa para criação de pipelines de dados em lote e streaming no AWS Glue 6.0. Com SDP, você define como devem ficar os dados usando SQL ou Python, e a estrutura determina automaticamente o plano de execução, resolve as dependências entre conjuntos de dados e executa ramificações independentes em paralelo.

O SDP simplifica o desenvolvimento de pipelines, eliminando o código clichê imperativo para leitura, gravação, registro de catálogos e ordenação de execuções. Você se concentra nas transformações de negócios enquanto a estrutura lida com a infraestrutura de pipeline.

O SDP está disponível apenas no AWS Glue versão 6.0 e posteriores.

Conceitos de SDP

Um pipeline consiste em um arquivo de manifesto YAML (spark-pipeline.yml) e um ou mais arquivos de transformação SQL ou Python. O SDP automaticamente:

Tipos de conjunto de dados

Três tipos de conjuntos de dados estão disponíveis no SDP:

Pré-requisitos

Para usar SDP, você precisa do seguinte:

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

Criar um pipeline

Para criar um pipeline SDP, conclua as seguintes etapas.

Etapa 1: criar o YAML do pipeline

Crie um arquivo chamado 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"

A tabela a seguir descreve os campos do YAML do pipeline.

Etapa 2: gravar transformações

Crie arquivos de transformação em um diretório transformations/. Você pode usar SQL, Python ou ambos no mesmo pipeline.

Exemplo em 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;

Exemplo em 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/")

Exemplo de tabela de streaming em 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/")
    )

Etapa 3: fazer upload no Amazon S3

Carregue os arquivos de pipeline no Amazon S3 como uma destas duas alternativas:

Etapa 4: criar e executar o trabalho no AWS Glue

Crie um trabalho do AWS Glue com os seguintes parâmetros:

O seguinte exemplo cria um trabalho de SDP usando a 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"
  }'

Executar pipelines

Você executa pipelines SDP usando o StartJobRun. Você pode controlar o comportamento da execução com argumentos de trabalho transmitidos durante a execução.

Modos de execução

Passe os seguintes argumentos para StartJobRun a fim de controlar a execução do pipeline:

Usar tabelas Iceberg com SDP

O Apache Iceberg é o formato de tabela recomendado para tabelas de streaming que exigem processamento incremental durável entre execuções, pois não depende do log de _spark_metadata baseado em arquivo usado pelo Hive nem das tabelas gerenciadas do AWS Glue. Você pode configurar o Iceberg com o SDP de duas maneiras, dependendo de desejar ou não que as tabelas de saída sejam registradas no Data Catalog.

Opção 1: Iceberg com o Data Catalog (recomendado)

Use essa opção quando quiser que as tabelas do Iceberg sejam registradas no Data Catalog (com table_type=ICEBERG) para que possam ser consultadas em outros mecanismos, como o Amazon Redshift e o Amazon EMR. Os dados e os metadados da tabela são armazenados no Amazon S3, e o estado incremental entre execuções é preservado. Adicione o seguinte à seção configuration do 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"

No YAML do pipeline, defina catalog: glue_catalog e defina database: como um banco de dados do AWS Glue. Ao criar o trabalho AWS Glue, defina --enable-spark-declarative-pipeline como true. Não configure o --enable-glue-datacatalog para tabelas Iceberg.

Opção 2: Iceberg com um catálogo baseado em arquivo (Hadoop)

Use essa opção quando não precisar do registro do Data Catalog. No Amazon S3, os metadados do Iceberg são baseados em arquivo e as tabelas não são registradas no Data Catalog. Adicione o seguinte à seção configuration do 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"

Use a opção 1 se precisar que as tabelas sejam registradas no Data Catalog e possam ser consultadas em outros mecanismos. Use a opção 2 apenas se não precisar de registro no Data Catalog.

Com o Iceberg configurado, você obtém os seguintes benefícios:

Você também pode ler incrementalmente uma tabela Iceberg como fonte de streaming. Em uma arquitetura medallion, uma tabela de streaming downstream pode consumir somente as linhas novas confirmadas em uma tabela Iceberg upstream em cada execução. O exemplo a seguir lê incrementalmente de uma tabela Iceberg bronze para uma tabela de streaming 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")

Considerações e limitações

Ao usar SDP, considere o seguinte:

Migração de scripts imperativos para o SDP

Você pode migrar os scripts imperativos existentes do Spark para o SDP de forma incremental:

O SDP pode referenciar qualquer tabela acessível por meio da SparkSession, incluindo tabelas existentes do Data Catalog, tabelas externas e referências entre bancos de dados.