Saluran Pipa Deklaratif Spark

Spark Declarative Pipelines (SDP) adalah kerangka kerja deklaratif untuk membangun pipeline data batch dan streaming di 6.0. AWS Glue Dengan SDP, Anda menentukan seperti apa data Anda seharusnya menggunakan SQL atau Python, dan kerangka kerja secara otomatis menentukan rencana eksekusi, menyelesaikan dependensi antara kumpulan data, dan menjalankan cabang independen secara paralel.

SDP menyederhanakan pengembangan pipeline dengan menghilangkan kode boilerplate imperatif untuk membaca, menulis, pendaftaran katalog, dan pemesanan eksekusi. Anda fokus pada transformasi bisnis sementara kerangka kerja menangani infrastruktur pipa.

SDP tersedia dalam AWS Glue versi 6.0 dan yang lebih baru.

Konsep SDP

Pipeline terdiri dari file manifes YAML (spark-pipeline.yml) dan satu atau lebih file transformasi SQL atau Python. SDP secara otomatis:

Jenis dataset

Tiga jenis dataset tersedia di SDP:

Prasyarat

Untuk menggunakan SDP, Anda memerlukan yang berikut:

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

Membuat pipa

Untuk membuat pipeline SDP, selesaikan langkah-langkah berikut.

Langkah 1: Buat pipa YAML

Buat file bernama 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"

Tabel berikut menjelaskan bidang YAML pipeline.

Langkah 2: Tulis transformasi

Buat file transformasi dalam transformations/ direktori. Anda dapat menggunakan SQL, Python, atau keduanya dalam pipeline yang sama.

Contoh 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;

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

Contoh tabel streaming 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/")
    )

Langkah 3: Unggah ke Amazon S3

Unggah file pipeline Anda ke Amazon S3 sebagai berikut:

Langkah 4: Buat dan jalankan AWS Glue pekerjaan

Buat AWS Glue pekerjaan dengan parameter berikut:

Contoh berikut membuat pekerjaan SDP menggunakan 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"
  }'

Menjalankan saluran pipa

Anda menjalankan pipeline SDP menggunakanStartJobRun. Anda dapat mengontrol perilaku eksekusi dengan argumen pekerjaan yang diteruskan pada waktu berjalan.

Mode Jalankan

Berikan argumen berikut StartJobRun untuk mengontrol eksekusi pipeline:

Menggunakan tabel Iceberg dengan SDP

Untuk tabel streaming yang memerlukan pemrosesan inkremental cross-run, gunakan Apache Iceberg. Hive-managed tabel streaming menyimpan metadata secara lokal dan tidak bertahan di seluruh proses pekerjaan.

Untuk mengkonfigurasi Iceberg, tambahkan yang berikut ini ke bagian spark-pipeline.yml konfigurasi Anda:

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"

Dengan Iceberg yang dikonfigurasi, Anda mendapatkan manfaat berikut:

Pertimbangan dan batasan

Pertimbangkan hal berikut saat Anda menggunakan SDP:

Migrasi dari skrip imperatif ke SDP

Anda dapat memigrasikan skrip Spark imperatif yang ada ke SDP secara bertahap:

SDP dapat mereferensikan tabel apa pun yang dapat diakses melalui SparkSession, termasuk tabel Katalog Data yang ada, tabel eksternal, dan referensi lintas database.