Spark 宣告管道

Spark 宣告管道 (SDP) 是用於在 AWS Glue 6.0 中建置批次和串流資料管道的宣告性架構。使用 SDP，您可以定義資料使用 SQL 或 Python 的外觀，而架構會自動決定執行計畫、解決資料集之間的相依性，並平行執行獨立分支。

SDP 透過消除讀取、寫入、目錄註冊和執行順序的強制性樣板程式碼，簡化管道開發。當架構處理管道基礎設施時，您會專注於業務轉型。

SDP 可在 6.0 版和更新 AWS Glue 版本中使用。

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 任務：

下列範例使用 CLI 建立 SDP AWS 任務：

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

執行中管道

您可以使用 執行 SDP 管道StartJobRun。您可以使用執行時間傳遞的任務引數來控制執行行為。

執行模式

將下列引數傳遞至 StartJobRun以控制管道執行：

搭配 SDP 使用 Iceberg 資料表

對於需要跨執行增量處理的串流資料表，請使用 Apache Iceberg。Hive 受管串流資料表會在本機存放中繼資料，而不會在任務執行期間保留。

若要設定 Iceberg，請將下列項目新增至您的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"

設定 Iceberg 後，您可以獲得下列優點：

考量和限制

當您使用 SDP 時，請考慮下列事項：

從命令式指令碼遷移至 SDP

您可以逐步將現有的必要 Spark 指令碼遷移至 SDP：

SDP 可以參考可透過 SparkSession 存取的任何資料表，包括現有的資料目錄資料表、外部資料表和跨資料庫參考。