태스크 유형과 의존성

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 레벨에서도 실패를 감지할 수 있도록 예외를 다시 던집니다

- task_key: "process_all_regions"
  for_each_task:
    inputs: '["us-east-1", "eu-west-1", "ap-northeast-2"]'
    concurrency: 3   # 3개 리전을 동시에 처리합니다
    task:
      task_key: "process_region"
      notebook_task:
        notebook_path: "/Workspace/etl/process_region"
        base_parameters:
          region: "{{input}}"   # 배열의 현재 요소가 주입됩니다

# discover_tenants 태스크: 처리할 테넌트 목록을 동적으로 생성합니다
import json

tenants = spark.table("catalog.meta.active_tenants").select("tenant_id").collect()
tenant_list = [row.tenant_id for row in tenants]

dbutils.jobs.taskValues.set(key="tenant_list", value=json.dumps(tenant_list))

- task_key: "process_tenants"
  depends_on:
    - task_key: "discover_tenants"
  for_each_task:
    inputs: "{{tasks.discover_tenants.values.tenant_list}}"
    concurrency: 5
    task:
      task_key: "process_single_tenant"
      python_wheel_task:
        package_name: "etl_pipeline"
        entry_point: "run_tenant"
        parameters: ["--tenant-id", "{{input}}"]

resources:
  jobs:
    daily_pipeline:
      name: "daily-sales-pipeline"
      tags:
        team: "data-engineering"
        project: "sales"
        environment: "${bundle.target}"

      parameters:
        - name: "target_date"
          default: ""
        - name: "environment"
          default: "production"

      tasks:
        - task_key: "ingest_orders"
          pipeline_task:
            pipeline_id: "${var.pipeline_id}"
          max_retries: 2
          min_retry_interval_millis: 60000
          timeout_seconds: 1800

        - task_key: "ingest_products"
          pipeline_task:
            pipeline_id: "${var.products_pipeline_id}"
          max_retries: 2

        - task_key: "transform"
          depends_on:
            - task_key: "ingest_orders"
            - task_key: "ingest_products"
          notebook_task:
            notebook_path: "/Workspace/etl/transform"
            base_parameters:
              date: "{{job.parameters.target_date}}"
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            num_workers: 4
            node_type_id: "i3.xlarge"
            aws_attributes:
              availability: "SPOT_WITH_FALLBACK"

        - task_key: "validate"
          depends_on:
            - task_key: "transform"
          sql_task:
            query:
              query_text: >
                SELECT COUNT(*) AS row_count
                FROM gold.daily_revenue
                WHERE sale_date = CURRENT_DATE()
            warehouse_id: "${var.warehouse_id}"

        - task_key: "notify"
          run_if: "ALL_DONE"
          depends_on:
            - task_key: "validate"
          notebook_task:
            notebook_path: "/Workspace/etl/send_notification"

      email_notifications:
        on_failure:
          - "data-team@company.com"
        on_success:
          - "data-team@company.com"

      webhook_notifications:
        on_failure:
          - id: "${var.slack_webhook_id}"

      schedule:
        quartz_cron_expression: "0 0 2 * * ?"
        timezone_id: "Asia/Seoul"

# 멱등성 보장: 덮어쓰기 모드 + 파티션 교체
spark.table("silver.orders") \
    .filter(f"sale_date = '{target_date}'") \
    .write \
    .mode("overwrite") \
    .option("replaceWhere", f"sale_date = '{target_date}'") \
    .saveAsTable("gold.daily_revenue")

# 잘못된 패턴: 모든 태스크 직렬화
ingest_orders → ingest_products → transform → validate → build_gold

# 올바른 패턴: 독립 태스크 병렬화
ingest_orders ──┐
                ├── transform → validate → build_gold
ingest_products─┘

# 잘못된 패턴: 대용량 데이터를 태스크 값으로 전달
rows = spark.table("silver.orders").collect()
dbutils.jobs.taskValues.set(key="data", value=str(rows))  # 크기 초과 위험

# 올바른 패턴: Delta 테이블 경로 또는 메타데이터만 전달
dbutils.jobs.taskValues.set(key="output_table", value="silver.orders")
dbutils.jobs.taskValues.set(key="row_count", value=spark.table("silver.orders").count())

- task_key: "long_running_transform"
  timeout_seconds: 7200   # 예상 1시간 → 2시간 타임아웃 설정
  max_retries: 1

# debugValue 없이 독립 실행 시 오류 발생
row_count = dbutils.jobs.taskValues.get(taskKey="validate", key="row_count")

# debugValue를 설정하면 독립 실행 시 해당 값을 사용합니다
row_count = dbutils.jobs.taskValues.get(
    taskKey="validate",
    key="row_count",
    default=0,
    debugValue=100   # 노트북 단독 실행 시 사용할 값
)