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 kumpulan data

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

Apache Iceberg adalah format tabel yang direkomendasikan untuk tabel streaming yang memerlukan pemrosesan inkremental cross-run yang tahan lama, karena tidak bergantung pada _spark_metadata log berbasis file yang digunakan oleh Hive atau tabel yang dikelola. AWS Glue Anda dapat mengonfigurasi Iceberg dengan SDP dalam dua cara, tergantung pada apakah Anda ingin tabel keluaran Anda terdaftar di Katalog Data.

Opsi 1: Gunung Es dengan Katalog Data (disarankan)

Gunakan opsi ini jika Anda ingin tabel Iceberg Anda terdaftar di Katalog Data (dengantable_type=ICEBERG) sehingga dapat dikueri dari mesin lain seperti, Amazon Redshift, dan Amazon EMR. Data tabel dan metadata disimpan di Amazon S3, dan status inkremental cross-run dipertahankan. Tambahkan yang berikut ini ke configuration bagian Andaspark-pipeline.yml:

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"

Di saluran YAML Anda, atur catalog: glue_catalog dan atur database: ke AWS Glue database. Saat Anda membuat AWS Glue pekerjaan, setel --enable-spark-declarative-pipeline ketrue. Jangan mengatur --enable-glue-datacatalog meja gunung es.

Opsi 2: Gunung es dengan katalog berbasis file (Hadoop)

Gunakan opsi ini jika Anda tidak memerlukan pendaftaran Katalog Data. Metadata gunung es berbasis file di Amazon S3, dan tabel tidak terdaftar di Katalog Data. Tambahkan yang berikut ini ke configuration bagian Andaspark-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"

Gunakan Opsi 1 jika Anda ingin tabel Anda terdaftar di Katalog Data dan dapat dikueri dari mesin lain. Gunakan Opsi 2 hanya jika Anda tidak memerlukan pendaftaran Katalog Data.

Dengan Iceberg dikonfigurasi, Anda mendapatkan manfaat berikut:

Anda juga dapat membaca secara bertahap dari tabel Iceberg sebagai sumber streaming. Dalam arsitektur medali, tabel streaming hilir hanya dapat menggunakan baris baru yang dikomitmen ke tabel Iceberg hulu pada setiap proses. Contoh berikut dibaca secara bertahap dari tabel Iceberg ke bronze tabel silver streaming:

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

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.