태스크 유형, 의존성, 실패 처리

ingest_orders ──┐
                ├── transform_orders ── validate_data ──┬── build_gold_tables ──┬── update_dashboard
ingest_products─┘                                       │                       └── notify_success
                                                        └── send_alert (실패 시)

- task_key: "route_by_quality"
  condition_task:
    op: "GREATER_THAN"
    left: "{{tasks.validate_data.values.row_count}}"
    right: "0"

- task_key: "build_gold_tables"
  depends_on:
    - task_key: "route_by_quality"
      outcome: "true"
  dbt_task:
    project_directory: "/dbt"
    commands: ["dbt run --select gold.*"]

- task_key: "send_alert"
  depends_on:
    - task_key: "route_by_quality"
      outcome: "false"
  notebook_task:
    notebook_path: "/Workspace/utils/alert"

# validate_data 태스크 (노트북)에서 행 수를 downstream으로 전달합니다
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

row_count = spark.table("silver.orders").filter("sale_date = current_date()").count()
error_count = spark.table("silver.orders").filter("status = 'ERROR'").count()

# 태스크 값 설정
dbutils.jobs.taskValues.set(key="row_count", value=row_count)
dbutils.jobs.taskValues.set(key="error_count", value=error_count)
dbutils.jobs.taskValues.set(key="validation_status", value="passed" if error_count == 0 else "failed")

print(f"row_count={row_count}, error_count={error_count}")

# build_gold_tables 태스크에서 upstream의 값을 읽습니다
row_count = dbutils.jobs.taskValues.get(
    taskKey="validate_data",   # upstream 태스크 key
    key="row_count",           # 읽을 값의 key
    default=0,                 # 값이 없을 경우 기본값
    debugValue=9999            # 노트북을 독립 실행할 때 사용할 디버그 값
)

validation_status = dbutils.jobs.taskValues.get(
    taskKey="validate_data",
    key="validation_status",
    default="unknown"
)

print(f"수신된 행 수: {row_count}, 검증 상태: {validation_status}")

if validation_status == "failed":
    raise Exception("Upstream validation failed. Aborting gold table build.")

- task_key: "notify_success"
  depends_on:
    - task_key: "build_gold_tables"
  notebook_task:
    notebook_path: "/Workspace/utils/notify"
    base_parameters:
      row_count: "{{tasks.validate_data.values.row_count}}"
      status: "{{tasks.validate_data.values.validation_status}}"

- task_key: "ingest_orders"
  pipeline_task:
    pipeline_id: "${var.pipeline_id}"
  max_retries: 3                # 최대 3회 재시도합니다
  min_retry_interval_millis: 60000   # 재시도 전 1분 대기합니다
  retry_on_timeout: true        # 타임아웃 시에도 재시도합니다
  timeout_seconds: 3600         # 1시간 후 타임아웃 처리합니다

# 실패 여부와 관계없이 항상 클린업을 실행하는 예시
- task_key: "cleanup"
  run_if: "ALL_DONE"
  depends_on:
    - task_key: "transform_orders"
  notebook_task:
    notebook_path: "/Workspace/utils/cleanup"

# transform_orders 태스크: 에러가 발생해도 빈 값을 설정하여 후속 태스크가 판단할 수 있게 합니다
try:
    df = spark.table("bronze.orders").transform(clean_orders)
    df.write.mode("overwrite").saveAsTable("silver.orders")
    dbutils.jobs.taskValues.set(key="transform_status", value="success")
    dbutils.jobs.taskValues.set(key="record_count", value=df.count())
except Exception as e:
    dbutils.jobs.taskValues.set(key="transform_status", value="failed")
    dbutils.jobs.taskValues.set(key="error_message", value=str(e))
    raise  # Job 레벨에서도 실패를 감지할 수 있도록 예외를 다시 던집니다