Spark 선언적 파이프라인

Spark 선언적 파이프라인(SDP)은 AWS Glue 6.0에서 배치 및 스트리밍 데이터 파이프라인을 빌드하기 위한 선언적 프레임워크입니다. SDP에서는 SQL 또는 Python을 사용하여 데이터의 모양을 정의할 수 있으며, 프레임워크는 실행 계획을 자동으로 결정하고, 데이터세트 간의 종속성을 해결하고, 독립 브랜치를 병렬로 실행합니다.

SDP는 읽기, 쓰기, 카탈로그 등록, 실행 순서를 위한 보일러플레이트 코드를 제거하여 파이프라인 개발을 간소화합니다. 프레임워크가 파이프라인 인프라를 처리하는 동안 사용자는 비즈니스 혁신에 집중할 수 있습니다.

SDP는 AWS Glue 버전 6.0 이상에서 사용할 수 있습니다.

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 작업을 생성합니다.

다음 예제에서는 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를 구성할 수 있습니다.

옵션 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는 기존 Data Catalog 테이블, 외부 테이블, 데이터베이스 간 참조를 포함하여 SparkSession을 통해 액세스할 수 있는 모든 테이블을 참조할 수 있습니다.