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 マネージドストリーミングテーブルは、メタデータをローカルに保存し、複数のジョブの実行をまたいで保持されることはありません。

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