Spark 宣言型パイプライン

Spark 宣言型パイプライン (SDP) は、AWS Glue 6.0 でバッチおよびストリーミングデータパイプラインを構築するための宣言型フレームワークです。SDP では、SQL または Python を使用してデータの外観を定義すると、フレームワークが自動的に実行プランを決定し、データセット間の依存関係を解決し、独立したブランチを並行実行します。

SDP は、読み取り、書き込み、カタログ登録、実行順序付けのための命令型のボイラープレートコードを排除することで、パイプライン開発を簡素化します。フレームワークがパイプラインインフラストラクチャに対処するため、ユーザーはビジネストランスフォーメーションに集中することができます。

SDP は、AWS Glue バージョン 6.0 以降で有効です。

SDP の概念

パイプラインは、YAML マニフェストファイル (spark-pipeline.yml) と、1 つ以上の SQL または Python 変換ファイルで構成されます。SDP は自動的に以下を行います。

データセットタイプ

SDP では、以下の 3 つのデータセットタイプを使用できます。

前提条件

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 に渡します。

SDP での Iceberg テーブルの使用

Apache Iceberg は、Hive または AWS Glue マネージドテーブルで使用されるファイルベースの _spark_metadata ログに依存しないため、耐久性のあるクロスラン増分処理を必要とするストリーミングテーブルに対して推奨されるテーブル形式です。SDP で Iceberg を設定する方法は 2 つあります。データカタログへの出力テーブルの登録の有無に応じて適切な方を使用してください。

オプション 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 を介してアクセス可能な任意のテーブルを参照できます。