파트 7: 골드 레이어 – Metrics, Watermarks, 및 Aggregations

Source: Dev.to

골드 테이블은 비즈니스 질문에 직접 답합니다

예시

• 지역별 시간당 여행 횟수
• 우편번호별 매출
• 시간대별 평균 거리

골드 테이블은:

• 집계된
• 최적화된
• 대시보드 준비 완료

이벤트 시간 및 워터마크 소개

골드 레이어가 지연된 데이터를 처리하도록 워터마크를 추가하고, 이벤트 시간을 기준으로 윈도잉을 사용해 집계를 올바르게 종료합니다. 아래 예시에서는 Spark가 열려 있는 1시간 윈도우에서 마지막으로 수신된 이벤트로부터 30분을 기다립니다. 워터마크 임계값(수신된 최대 tpep_pickup_datetime – 30분)이 윈도우 종료 시간을 초과하면 윈도우가 닫히고 집계 결과가 최종 확정됩니다.

from pyspark.sql.functions import *

silver_df = spark.readStream.format("delta").table("nyc_taxi.silver.taxi_trips_enriched")

gold_df = (
    silver_df
        .withWatermark("tpep_pickup_datetime", "30 minutes")
        .groupBy(
            window("tpep_pickup_datetime", "1 hour"),
            "region"
        )
        .agg(
            count("*").alias("trip_count"),
            sum("fare_amount").alias("total_fare"),
            avg("trip_distance").alias("avg_distance")
        )
)

# Stream the result to a gold Delta table
(
    gold_df.writeStream
        .option('mergeSchema', 'true')
        .trigger(availableNow=True)
        .option("checkpointLocation", "/Volumes/nyc_taxi/infra/checkpoints/gold/taxi_metrics")
        .outputMode("append")
        .toTable("nyc_taxi.gold.taxi_metrics")
)

상세 택시 여행 메트릭 뷰

골드 레이어는 여러 뷰를 제공할 수 있습니다. 아래 예시는 상세 택시 여행 메트릭을 강조하는 뷰를 생성합니다.

from pyspark.sql.functions import *

silver_stream = spark.readStream.format("delta").table("nyc_taxi.silver.taxi_trips_enriched")

gold_stream = (
    silver_stream
        .withWatermark("tpep_pickup_datetime", "30 minutes")
        .withColumn("trip_date", to_date("tpep_pickup_datetime"))
        .withColumn("trip_hour", hour("tpep_pickup_datetime"))
        .groupBy(
            window("tpep_pickup_datetime", "1 hour"),
            "trip_date",
            "trip_hour",
            "pickup_zip",
            "region"
        )
        .agg(
            count("*").alias("total_trips"),
            sum("fare_amount").alias("total_revenue"),
            avg("fare_amount").alias("avg_fare"),
            avg("trip_distance").alias("avg_distance")
        )
)

gold_stream.writeStream \
    .format("delta") \
    .trigger(availableNow=True) \
    .option("checkpointLocation", "/Volumes/nyc_taxi/infra/checkpoints/gold/taxi_trip_metrics") \
    .outputMode("append") \
    .table("nyc_taxi.gold.taxi_trip_metrics")

데이터가 이제 집계되어 골드 Delta 테이블에 저장되었으며, 비즈니스 인사이트를 위한 준비가 완료되었습니다.

학습을 즐기세요!