Spark 声明式管道

Spark 声明式管道（SDP）是一个声明式框架，用于在 AWS Glue 6.0 中构建批处理和流式传输数据管道。使用 SDP，您可以通过 SQL 或 Python 定义数据的目标形态，而框架将自动确定执行计划、解析数据集之间的依赖关系，并行运行相互独立的分支。

SDP 省去了用于读取、写入、目录注册和执行排序的命令式样板代码，从而简化了管道开发过程。您可以专注于业务转换，而管道基础设施则由框架负责处理。

SDP 在 AWS Glue 6.0 及更高版本中可用。

SDP 概念

管道由一个 YAML 清单文件（spark-pipeline.yml）和一个或多个 SQL 或 Python 转换文件组成。SDP 自动执行以下操作：

数据集类型

SDP 中提供三种数据集类型：

先决条件

要使用 SDP，您需要以下内容：

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

创建管道

要创建 SDP 管道，请完成以下步骤。

步骤 1：创建管道 YAML

创建一个名为 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"

下表描述了管道 YAML 字段。

步骤 2：编写转换

在 transformations/ 目录中创建转换文件。您可以在同一个管道中使用 SQL、Python 或此两者。

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;

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

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

步骤 3：上传到 Amazon S3

将您的管道文件上传到 Amazon S3 作为以下任一内容：

步骤 4：创建并运行 AWS Glue 作业

使用以下参数创建 AWS Glue 作业：

以下示例使用 AWS CLI 创建 SDP 作业：

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

运行管道

您可以使用 StartJobRun 运行 SDP 管道。您可以使用在运行时传递的作业参数来控制执行行为。

运行模式

将以下参数传递给 StartJobRun 以控制管道执行：

将 Iceberg 表与 SDP 结合使用

对于需要持久、跨运行增量处理的流式传输表，Apache Iceberg 是推荐的表格式，因为它不依赖于 Hive 或 AWS Glue 托管表所使用的基于文件的 _spark_metadata 日志。您可以通过两种方式使用 SDP 配置 Iceberg，具体取决于是否要在数据目录中注册输出表。

选项 1：使用数据目录的 Iceberg（推荐）

如果您希望将 Iceberg 表注册到数据目录中（使用 table_type=ICEBERG），以便通过 Amazon Redshift 和 Amazon EMR 等其他引擎查询这些表，请使用此选项。表数据和元数据存储在 Amazon S3 中，并保留跨运行增量状态。将以下内容添加到您 spark-pipeline.yml 的 configuration 部分：

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"

在您的管道 YAML 中，设置 catalog: glue_catalog 并将 database: 设置为 AWS Glue 数据库。创建 AWS Glue 作业时，将 --enable-spark-declarative-pipeline 设置为 true。请勿为 Iceberg 表设置 --enable-glue-datacatalog。

选项 2：使用基于文件的（Hadoop）目录的 Iceberg

当您不需要数据目录注册时，请使用此选项。Iceberg 元数据在 Amazon S3 中是基于文件的，并且这些表未在数据目录中注册。将以下内容添加到您 spark-pipeline.yml 的 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"

如果您需要在数据目录中注册表并可从其他引擎查询，请使用选项 1。仅当不需要注册数据目录时，才使用选项 2。

配置 Iceberg 后，您可以获得以下优势：

您也可以将 Iceberg 表作为流式数据源进行增量读取。在奖章架构中，下游流式传输表在每次运行时只能处理新提交到上游 Iceberg 表的行。以下示例以增量方式将数据从 Iceberg bronze 表读取到 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")

注意事项和限制

在使用 SDP 时，请考虑以下几点：

从命令式脚本迁移到 SDP

您可以将现有的命令式 Spark 脚本以增量方式迁移到 SDP：

SDP 可以引用任何可通过 SparkSession 访问的表，包括现有的 Data Catalog 表、外部表以及跨数据库引用。