Databricks Jobs

Lakeflow Jobs Development

FIRST: Use the parent databricks-core skill for CLI basics, authentication, profile selection, and data exploration commands.

Lakeflow Jobs orchestrate data workflows with multi-task DAGs, flexible triggers, and comprehensive monitoring. Jobs support diverse task types and can be managed via Asset Bundles (DABs), Python SDK, or CLI.

Reference Files

Scaffolding a New Job Project

Use databricks bundle init with a config file to scaffold non-interactively. This creates a project in the <project_name>/ directory:

databricks bundle init default-python --config-file <(echo '{"project_name": "my_job", "include_job": "yes", "include_pipeline": "no", "include_python": "yes", "serverless": "yes"}') --profile <PROFILE> < /dev/null

• project_name: letters, numbers, underscores only

After scaffolding, create CLAUDE.md and AGENTS.md in the project directory. These files are essential to provide agents with guidance on how to work with the project. Use this content:

# Declarative Automation Bundles Project

This project uses Declarative Automation Bundles (formerly Databricks Asset Bundles) for deployment.

## Prerequisites

Install the Databricks CLI (>= v0.288.0) if not already installed:
- macOS: `brew tap databricks/tap && brew install databricks`
- Linux: `curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh`
- Windows: `winget install Databricks.DatabricksCLI`

Verify: `databricks -v`

## For AI Agents

Read the `databricks-core` skill for CLI basics, authentication, and deployment workflow.
Read the `databricks-jobs` skill for job-specific guidance.

If skills are not available, install them: `databricks aitools install`

Project Structure

my-job-project/
├── databricks.yml              # Bundle configuration
├── resources/
│   └── my_job.job.yml          # Job definition
├── src/
│   ├── my_notebook.ipynb       # Notebook tasks
│   └── my_module/              # Python wheel package
│       ├── __init__.py
│       └── main.py
├── tests/
│   └── test_main.py
└── pyproject.toml              # Python project config (if using wheels)

Quick Start

Asset Bundles (DABs) — recommended

# resources/jobs.yml
resources:
  jobs:
    my_etl_job:
      name: "[${bundle.target}] My ETL Job"
      tasks:
        - task_key: extract
          notebook_task:
            notebook_path: ../src/notebooks/extract.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Task, NotebookTask, Source

w = WorkspaceClient()

job = w.jobs.create(
    name="my-etl-job",
    tasks=[
        Task(
            task_key="extract",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/Shared/etl/extract",
                source=Source.WORKSPACE,
            ),
        ),
    ],
)
print(f"Created job: {job.job_id}")

CLI

databricks jobs create --json '{
  "name": "my-etl-job",
  "tasks": [{
    "task_key": "extract",
    "notebook_task": {
      "notebook_path": "/Workspace/Shared/etl/extract",
      "source": "WORKSPACE"
    }
  }]
}'

Core Concepts

Multi-Task Workflows

Jobs support DAG-based task dependencies:

tasks:
  - task_key: extract
    notebook_task:
      notebook_path: ../src/extract.py

  - task_key: transform
    depends_on:
      - task_key: extract
    notebook_task:
      notebook_path: ../src/transform.py

  - task_key: load
    depends_on:
      - task_key: transform
    run_if: ALL_SUCCESS  # Only run if all dependencies succeed
    notebook_task:
      notebook_path: ../src/load.py

run_if conditions:

• ALL_SUCCESS (default) — run when all dependencies succeed
• ALL_DONE — run when all dependencies complete (success or failure)
• AT_LEAST_ONE_SUCCESS — run when at least one dependency succeeds
• NONE_FAILED — run when no dependencies failed
• ALL_FAILED — run when all dependencies failed
• AT_LEAST_ONE_FAILED — run when at least one dependency failed

Task Types Summary

Trigger Types Summary

Compute Configuration

Job Clusters (recommended)

Define reusable cluster configurations shared across tasks:

job_clusters:
  - job_cluster_key: shared_cluster
    new_cluster:
      spark_version: "15.4.x-scala2.12"
      node_type_id: "i3.xlarge"
      num_workers: 2
      spark_conf:
        spark.speculation: "true"

tasks:
  - task_key: my_task
    job_cluster_key: shared_cluster
    notebook_task:
      notebook_path: ../src/notebook.py

Autoscaling Clusters

new_cluster:
  spark_version: "15.4.x-scala2.12"
  node_type_id: "i3.xlarge"
  autoscale:
    min_workers: 2
    max_workers: 8

Existing Cluster

tasks:
  - task_key: my_task
    existing_cluster_id: "0123-456789-abcdef12"
    notebook_task:
      notebook_path: ../src/notebook.py

Serverless Compute

For notebook and Python tasks, omit cluster configuration to use serverless:

tasks:
  - task_key: serverless_task
    notebook_task:
      notebook_path: ../src/notebook.py
    # No cluster config = serverless

Job Parameters

Parameters defined at job level are passed to ALL tasks (no need to repeat per task):

parameters:
  - name: env
    default: "dev"
  - name: date
    default: "{{start_date}}"  # Dynamic value reference

Access in notebooks:

catalog = dbutils.widgets.get("env")
load_date = dbutils.widgets.get("date")

Pass to specific tasks:

tasks:
  - task_key: my_task
    notebook_task:
      notebook_path: ../src/notebook.py
      base_parameters:
        env: "{{job.parameters.env}}"
        custom_param: "value"

Common Operations

Python SDK

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# List jobs
jobs = w.jobs.list()

# Get job details
job = w.jobs.get(job_id=12345)

# Run job now
run = w.jobs.run_now(job_id=12345)

# Run with parameters
run = w.jobs.run_now(
    job_id=12345,
    job_parameters={"env": "prod", "date": "2024-01-15"},
)

# Cancel run
w.jobs.cancel_run(run_id=run.run_id)

# Delete job
w.jobs.delete(job_id=12345)

CLI

# List jobs
databricks jobs list

# Get job details
databricks jobs get 12345

# Run job
databricks jobs run-now 12345

# Run with parameters (must use --json with job_id inside)
databricks jobs run-now --json '{"job_id": 12345, "job_parameters": {"env": "prod"}}'

# Cancel run
databricks jobs cancel-run 67890

# Delete job
databricks jobs delete 12345

Asset Bundle Operations

# Validate configuration
databricks bundle validate --profile <profile>

# Deploy to a target
databricks bundle deploy -t dev --profile <profile>

# Run a job
databricks bundle run <job_name> -t dev --profile <profile>

# Check run status
databricks jobs get-run --run-id <id> --profile <profile>

# Destroy resources
databricks bundle destroy --auto-approve

Permissions (DABs)

resources:
  jobs:
    my_job:
      name: "My Job"
      permissions:
        - level: CAN_VIEW
          group_name: "data-analysts"
        - level: CAN_MANAGE_RUN
          group_name: "data-engineers"
        - level: CAN_MANAGE
          user_name: "admin@example.com"

Permission levels:

• CAN_VIEW — view job and run history
• CAN_MANAGE_RUN — view, trigger, and cancel runs
• CAN_MANAGE — full control including edit and delete

Unit Testing

Run unit tests locally:

uv run pytest

Development Workflow

1. Validate: databricks bundle validate --profile <profile> 2. Deploy: databricks bundle deploy -t dev --profile <profile> 3. Run: databricks bundle run <job_name> -t dev --profile <profile> 4. Check run status: databricks jobs get-run --run-id <id> --profile <profile>

Common Issues

Related Skills

• databricks-dabs — DABs configuration patterns shared by jobs and pipelines
• databricks-pipelines — SDP (formerly DLT) pipelines triggered by pipeline_task

Documentation

• Lakeflow Jobs
• Task types
• Declarative Automation Bundles
• Jobs API Reference
• Bundle Examples Repository

Complete Examples

Contents

• ETL Pipeline with Multiple Tasks
• Scheduled Data Warehouse Refresh
• Event-Driven Pipeline
• ML Training Pipeline
• Multi-Environment Deployment
• Streaming Job
• Cross-Job Orchestration

---

ETL Pipeline with Multiple Tasks

A classic extract-transform-load pipeline with task dependencies.

DABs YAML

# resources/etl_job.yml
resources:
  jobs:
    daily_etl:
      name: "[${bundle.target}] Daily ETL Pipeline"

      # Schedule: Daily at 6 AM UTC
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "UTC"
        pause_status: UNPAUSED

      # Job parameters
      parameters:
        - name: load_date
          default: "{{start_date}}"
        - name: env
          default: "${bundle.target}"

      # Shared cluster for all tasks
      job_clusters:
        - job_cluster_key: etl_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 4
            spark_conf:
              spark.sql.shuffle.partitions: "200"

      # Email notifications
      email_notifications:
        on_failure:
          - "data-team@example.com"
        on_success:
          - "data-team@example.com"

      tasks:
        # Extract from source systems
        - task_key: extract_orders
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/extract_orders.py
            base_parameters:
              load_date: "{{job.parameters.load_date}}"

        - task_key: extract_customers
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/extract_customers.py
            base_parameters:
              load_date: "{{job.parameters.load_date}}"

        - task_key: extract_products
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/extract_products.py

        # Transform: wait for all extracts
        - task_key: transform_facts
          depends_on:
            - task_key: extract_orders
            - task_key: extract_customers
            - task_key: extract_products
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/transform_facts.py
            base_parameters:
              load_date: "{{job.parameters.load_date}}"

        # Load: run after transform
        - task_key: load_warehouse
          depends_on:
            - task_key: transform_facts
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/load_warehouse.py

        # Data quality check
        - task_key: validate_data
          depends_on:
            - task_key: load_warehouse
          run_if: ALL_SUCCESS
          job_cluster_key: etl_cluster
          notebook_task:
            notebook_path: ../src/notebooks/validate_data.py

      permissions:
        - level: CAN_VIEW
          group_name: "data-analysts"
        - level: CAN_MANAGE_RUN
          group_name: "data-engineers"

Python SDK Equivalent

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import (
    Task, NotebookTask, Source,
    JobCluster, ClusterSpec,
    CronSchedule, PauseStatus,
    JobEmailNotifications,
    JobParameterDefinition
)

w = WorkspaceClient()

job = w.jobs.create(
    name="Daily ETL Pipeline",
    schedule=CronSchedule(
        quartz_cron_expression="0 0 6 * * ?",
        timezone_id="UTC",
        pause_status=PauseStatus.UNPAUSED
    ),
    parameters=[
        JobParameterDefinition(name="load_date", default="{{start_date}}"),
        JobParameterDefinition(name="env", default="prod")
    ],
    job_clusters=[
        JobCluster(
            job_cluster_key="etl_cluster",
            new_cluster=ClusterSpec(
                spark_version="15.4.x-scala2.12",
                node_type_id="i3.xlarge",
                num_workers=4
            )
        )
    ],
    email_notifications=JobEmailNotifications(
        on_failure=["data-team@example.com"],
        on_success=["data-team@example.com"]
    ),
    tasks=[
        Task(
            task_key="extract_orders",
            job_cluster_key="etl_cluster",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/etl/extract_orders",
                source=Source.WORKSPACE,
                base_parameters={"load_date": "{{job.parameters.load_date}}"}
            )
        ),
        Task(
            task_key="extract_customers",
            job_cluster_key="etl_cluster",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/etl/extract_customers",
                source=Source.WORKSPACE
            )
        ),
        Task(
            task_key="transform_facts",
            depends_on=[
                {"task_key": "extract_orders"},
                {"task_key": "extract_customers"}
            ],
            job_cluster_key="etl_cluster",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/etl/transform_facts",
                source=Source.WORKSPACE
            )
        ),
        Task(
            task_key="load_warehouse",
            depends_on=[{"task_key": "transform_facts"}],
            job_cluster_key="etl_cluster",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/etl/load_warehouse",
                source=Source.WORKSPACE
            )
        )
    ]
)

print(f"Created job: {job.job_id}")

---

Scheduled Data Warehouse Refresh

SQL-based warehouse refresh with multiple queries.

DABs YAML

resources:
  jobs:
    warehouse_refresh:
      name: "[${bundle.target}] Warehouse Refresh"

      schedule:
        quartz_cron_expression: "0 0 4 * * ?"  # 4 AM daily
        timezone_id: "America/New_York"
        pause_status: UNPAUSED

      tasks:
        # Refresh dimension tables
        - task_key: refresh_dim_customers
          sql_task:
            file:
              path: ../src/sql/refresh_dim_customers.sql
              source: WORKSPACE
            warehouse_id: ${var.warehouse_id}

        - task_key: refresh_dim_products
          sql_task:
            file:
              path: ../src/sql/refresh_dim_products.sql
              source: WORKSPACE
            warehouse_id: ${var.warehouse_id}

        # Refresh fact tables (depends on dimensions)
        - task_key: refresh_fact_sales
          depends_on:
            - task_key: refresh_dim_customers
            - task_key: refresh_dim_products
          sql_task:
            file:
              path: ../src/sql/refresh_fact_sales.sql
              source: WORKSPACE
            warehouse_id: ${var.warehouse_id}

        # Update aggregations
        - task_key: update_aggregations
          depends_on:
            - task_key: refresh_fact_sales
          sql_task:
            file:
              path: ../src/sql/update_aggregations.sql
              source: WORKSPACE
            warehouse_id: ${var.warehouse_id}

        # Refresh dashboard
        - task_key: refresh_dashboard
          depends_on:
            - task_key: update_aggregations
          sql_task:
            dashboard:
              dashboard_id: "dashboard-uuid-here"
            warehouse_id: ${var.warehouse_id}

---

Event-Driven Pipeline

Pipeline triggered by file arrival and table updates.

DABs YAML

resources:
  jobs:
    event_driven_pipeline:
      name: "[${bundle.target}] Event-Driven Pipeline"

      # Trigger on file arrival
      trigger:
        pause_status: UNPAUSED
        file_arrival:
          url: "s3://data-lake/incoming/orders/"
          min_time_between_triggers_seconds: 300  # 5 min cooldown
          wait_after_last_change_seconds: 60  # Wait for batch completion

      # Health monitoring
      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 1800  # Alert if > 30 min

      email_notifications:
        on_failure:
          - "data-alerts@example.com"
        on_duration_warning_threshold_exceeded:
          - "data-alerts@example.com"

      tasks:
        - task_key: process_incoming
          notebook_task:
            notebook_path: ../src/notebooks/process_incoming_files.py
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            autoscale:
              min_workers: 2
              max_workers: 10

Table Update Trigger Example

resources:
  jobs:
    table_triggered_job:
      name: "[${bundle.target}] Table Update Handler"

      trigger:
        pause_status: UNPAUSED
        table_update:
          table_names:
            - "main.bronze.raw_orders"
            - "main.bronze.raw_inventory"
          condition: ANY_UPDATED
          min_time_between_triggers_seconds: 600
          wait_after_last_change_seconds: 120

      tasks:
        - task_key: process_updates
          notebook_task:
            notebook_path: ../src/notebooks/process_table_updates.py

---

ML Training Pipeline

Machine learning workflow with training, evaluation, and deployment.

DABs YAML

resources:
  jobs:
    ml_training:
      name: "[${bundle.target}] ML Training Pipeline"

      # Weekly retraining
      schedule:
        quartz_cron_expression: "0 0 2 ? * SUN"  # Sunday 2 AM
        timezone_id: "UTC"
        pause_status: UNPAUSED

      parameters:
        - name: model_name
          default: "sales_forecaster"
        - name: experiment_name
          default: "/Shared/experiments/sales_forecast"

      # GPU cluster for training
      job_clusters:
        - job_cluster_key: gpu_cluster
          new_cluster:
            spark_version: "15.4.x-gpu-ml-scala2.12"
            node_type_id: "g5.xlarge"
            num_workers: 2
            aws_attributes:
              first_on_demand: 1

        - job_cluster_key: cpu_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 4

      # ML environment
      environments:
        - environment_key: ml_env
          spec:
            dependencies:
              - mlflow>=2.10.0
              - scikit-learn>=1.4.0
              - pandas>=2.0.0
              - xgboost>=2.0.0

      tasks:
        # Data preparation
        - task_key: prepare_training_data
          job_cluster_key: cpu_cluster
          environment_key: ml_env
          notebook_task:
            notebook_path: ../src/ml/prepare_training_data.py
            base_parameters:
              output_table: "main.ml.training_data"

        # Feature engineering
        - task_key: engineer_features
          depends_on:
            - task_key: prepare_training_data
          job_cluster_key: cpu_cluster
          environment_key: ml_env
          notebook_task:
            notebook_path: ../src/ml/engineer_features.py

        # Model training
        - task_key: train_model
          depends_on:
            - task_key: engineer_features
          job_cluster_key: gpu_cluster
          environment_key: ml_env
          notebook_task:
            notebook_path: ../src/ml/train_model.py
            base_parameters:
              model_name: "{{job.parameters.model_name}}"
              experiment_name: "{{job.parameters.experiment_name}}"

        # Model evaluation
        - task_key: evaluate_model
          depends_on:
            - task_key: train_model
          job_cluster_key: cpu_cluster
          environment_key: ml_env
          notebook_task:
            notebook_path: ../src/ml/evaluate_model.py

        # Conditional deployment (only on success)
        - task_key: deploy_model
          depends_on:
            - task_key: evaluate_model
          run_if: ALL_SUCCESS
          job_cluster_key: cpu_cluster
          environment_key: ml_env
          notebook_task:
            notebook_path: ../src/ml/deploy_model.py
            base_parameters:
              model_name: "{{job.parameters.model_name}}"

---

Multi-Environment Deployment

Job configuration with environment-specific settings.

databricks.yml

bundle:
  name: data-pipeline

include:
  - resources/*.yml

variables:
  warehouse_id:
    lookup:
      warehouse: "Shared SQL Warehouse"
  notification_email:
    default: "data-team@example.com"

targets:
  dev:
    default: true
    mode: development
    workspace:
      profile: dev-profile
    variables:
      notification_email: "dev-team@example.com"

  staging:
    mode: development
    workspace:
      profile: staging-profile

  prod:
    mode: production
    workspace:
      profile: prod-profile
    run_as:
      service_principal_name: "production-sp"

resources/jobs.yml

resources:
  jobs:
    data_pipeline:
      name: "[${bundle.target}] Data Pipeline"

      # Only schedule in prod
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "UTC"
        pause_status: ${if(bundle.target == "prod", "UNPAUSED", "PAUSED")}

      # Environment-specific cluster sizing
      job_clusters:
        - job_cluster_key: main_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: ${if(bundle.target == "prod", "i3.2xlarge", "i3.xlarge")}
            num_workers: ${if(bundle.target == "prod", 8, 2)}

      email_notifications:
        on_failure:
          - ${var.notification_email}

      tasks:
        - task_key: process_data
          job_cluster_key: main_cluster
          notebook_task:
            notebook_path: ../src/notebooks/process_data.py
            base_parameters:
              env: "${bundle.target}"
              catalog: "${bundle.target}_catalog"

      permissions:
        - level: CAN_VIEW
          group_name: "data-analysts"
        - level: CAN_MANAGE_RUN
          group_name: "data-engineers"
        - level: CAN_MANAGE
          service_principal_name: "deployment-sp"

---

Streaming Job

Continuous streaming job with monitoring.

DABs YAML

resources:
  jobs:
    streaming_processor:
      name: "[${bundle.target}] Streaming Processor"

      # Continuous execution
      continuous:
        pause_status: UNPAUSED

      # Health monitoring for streaming
      health:
        rules:
          - metric: STREAMING_BACKLOG_SECONDS
            op: GREATER_THAN
            value: 300  # Alert if > 5 min behind
          - metric: STREAMING_BACKLOG_RECORDS
            op: GREATER_THAN
            value: 1000000  # Alert if > 1M records behind

      email_notifications:
        on_failure:
          - "streaming-alerts@example.com"
        on_streaming_backlog_exceeded:
          - "streaming-alerts@example.com"

      webhook_notifications:
        on_failure:
          - id: "pagerduty-streaming-alerts"
        on_streaming_backlog_exceeded:
          - id: "slack-streaming-channel"

      tasks:
        - task_key: stream_processor
          notebook_task:
            notebook_path: ../src/notebooks/stream_processor.py
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            autoscale:
              min_workers: 2
              max_workers: 16
            spark_conf:
              spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled: "true"
              spark.sql.streaming.stateStore.providerClass: "com.databricks.sql.streaming.state.RocksDBStateStoreProvider"

---

Cross-Job Orchestration

Multiple jobs with dependencies using run_job_task.

DABs YAML

resources:
  jobs:
    # Data ingestion job
    ingestion_job:
      name: "[${bundle.target}] Data Ingestion"
      tasks:
        - task_key: ingest
          notebook_task:
            notebook_path: ../src/notebooks/ingest.py

    # Data transformation job
    transformation_job:
      name: "[${bundle.target}] Data Transformation"
      tasks:
        - task_key: transform
          notebook_task:
            notebook_path: ../src/notebooks/transform.py

    # Master orchestration job
    orchestrator:
      name: "[${bundle.target}] Master Orchestrator"

      schedule:
        quartz_cron_expression: "0 0 1 * * ?"
        timezone_id: "UTC"
        pause_status: UNPAUSED

      tasks:
        # Run ingestion first
        - task_key: run_ingestion
          run_job_task:
            job_id: ${resources.jobs.ingestion_job.id}

        # Run transformation after ingestion
        - task_key: run_transformation
          depends_on:
            - task_key: run_ingestion
          run_job_task:
            job_id: ${resources.jobs.transformation_job.id}

        # Final validation
        - task_key: validate_all
          depends_on:
            - task_key: run_transformation
          notebook_task:
            notebook_path: ../src/notebooks/validate_all.py

---

For Each Task - Parallel Processing

Process multiple items in parallel using for_each_task.

DABs YAML

resources:
  jobs:
    parallel_processor:
      name: "[${bundle.target}] Parallel Region Processor"

      schedule:
        quartz_cron_expression: "0 0 8 * * ?"
        timezone_id: "UTC"
        pause_status: UNPAUSED

      tasks:
        # Generate list of items to process
        - task_key: get_regions
          notebook_task:
            notebook_path: ../src/notebooks/get_active_regions.py

        # Process each region in parallel
        - task_key: process_regions
          depends_on:
            - task_key: get_regions
          for_each_task:
            inputs: "{{tasks.get_regions.values.regions}}"
            concurrency: 10  # Max 10 parallel
            task:
              task_key: process_region
              notebook_task:
                notebook_path: ../src/notebooks/process_region.py
                base_parameters:
                  region: "{{input}}"

        # Aggregate results after all regions processed
        - task_key: aggregate_results
          depends_on:
            - task_key: process_regions
          run_if: ALL_DONE  # Run even if some regions failed
          notebook_task:
            notebook_path: ../src/notebooks/aggregate_results.py

Notebook: get_active_regions.py

# Get list of active regions to process
regions = spark.sql("""
    SELECT DISTINCT region_code
    FROM main.config.active_regions
    WHERE is_active = true
""").collect()

region_list = [row.region_code for row in regions]

# Pass to downstream for_each_task
dbutils.jobs.taskValues.set(key="regions", value=region_list)

Notebook: process_region.py

# Get region from parameter
region = dbutils.widgets.get("region")

# Process data for this region
df = spark.sql(f"""
    SELECT * FROM main.bronze.orders
    WHERE region = '{region}'
""")

# Transform and write
df_transformed = transform_orders(df)
df_transformed.write.mode("append").saveAsTable(f"main.silver.orders_{region}")

print(f"Processed region: {region}")

Notifications and Monitoring Reference

Contents

• Email Notifications
• Webhook Notifications
• Health Rules
• Timeout Configuration
• Retry Configuration
• Run Queue Settings

---

Email Notifications

Send email alerts for job lifecycle events.

DABs YAML

resources:
  jobs:
    monitored_job:
      name: "Monitored Job"
      email_notifications:
        on_start:
          - "team@example.com"
        on_success:
          - "team@example.com"
        on_failure:
          - "oncall@example.com"
          - "team@example.com"
        on_duration_warning_threshold_exceeded:
          - "oncall@example.com"
        no_alert_for_skipped_runs: true
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import JobEmailNotifications

w = WorkspaceClient()

job = w.jobs.create(
    name="Monitored Job",
    email_notifications=JobEmailNotifications(
        on_start=["team@example.com"],
        on_success=["team@example.com"],
        on_failure=["oncall@example.com", "team@example.com"],
        on_duration_warning_threshold_exceeded=["oncall@example.com"],
        no_alert_for_skipped_runs=True
    ),
    tasks=[...]
)

Email Notification Events

Task-Level Email Notifications

tasks:
  - task_key: critical_task
    email_notifications:
      on_start:
        - "task-owner@example.com"
      on_success:
        - "task-owner@example.com"
      on_failure:
        - "oncall@example.com"
    notebook_task:
      notebook_path: ../src/critical.py

---

Webhook Notifications

Send HTTP webhooks for job events (Slack, PagerDuty, custom endpoints).

Create Notification Destination First

Before using webhooks, create a notification destination in the workspace:

Python SDK:

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.settings import (
    CreateNotificationDestinationRequest,
    DestinationType,
    SlackConfig
)

w = WorkspaceClient()

# Create Slack destination
destination = w.notification_destinations.create(
    display_name="Slack Alerts",
    config=SlackConfig(
        url="https://hooks.slack.com/services/XXX/YYY/ZZZ"
    )
)

print(f"Destination ID: {destination.id}")

DABs YAML

resources:
  jobs:
    webhook_job:
      name: "Job with Webhooks"
      webhook_notifications:
        on_start:
          - id: "notification-destination-uuid"
        on_success:
          - id: "notification-destination-uuid"
        on_failure:
          - id: "pagerduty-destination-uuid"
        on_duration_warning_threshold_exceeded:
          - id: "slack-destination-uuid"
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import WebhookNotifications, Webhook

w = WorkspaceClient()

job = w.jobs.create(
    name="Job with Webhooks",
    webhook_notifications=WebhookNotifications(
        on_start=[Webhook(id="notification-destination-uuid")],
        on_success=[Webhook(id="notification-destination-uuid")],
        on_failure=[Webhook(id="pagerduty-destination-uuid")],
        on_duration_warning_threshold_exceeded=[Webhook(id="slack-destination-uuid")]
    ),
    tasks=[...]
)

Supported Destinations

Task-Level Webhooks

tasks:
  - task_key: critical_task
    webhook_notifications:
      on_failure:
        - id: "pagerduty-destination-uuid"
    notebook_task:
      notebook_path: ../src/critical.py

---

Health Rules

Monitor job health metrics and trigger alerts.

DABs YAML

resources:
  jobs:
    health_monitored:
      name: "Health Monitored Job"
      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 3600  # Alert if run > 1 hour
          - metric: STREAMING_BACKLOG_BYTES
            op: GREATER_THAN
            value: 1073741824  # Alert if backlog > 1GB
          - metric: STREAMING_BACKLOG_SECONDS
            op: GREATER_THAN
            value: 300  # Alert if backlog > 5 minutes
          - metric: STREAMING_BACKLOG_FILES
            op: GREATER_THAN
            value: 1000  # Alert if backlog > 1000 files
          - metric: STREAMING_BACKLOG_RECORDS
            op: GREATER_THAN
            value: 100000  # Alert if backlog > 100k records
      email_notifications:
        on_duration_warning_threshold_exceeded:
          - "oncall@example.com"
        on_streaming_backlog_exceeded:
          - "oncall@example.com"
      tasks:
        - task_key: streaming
          notebook_task:
            notebook_path: ../src/streaming.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import JobsHealthRules, JobsHealthRule, JobsHealthMetric, JobsHealthOperator

w = WorkspaceClient()

job = w.jobs.create(
    name="Health Monitored Job",
    health=JobsHealthRules(
        rules=[
            JobsHealthRule(
                metric=JobsHealthMetric.RUN_DURATION_SECONDS,
                op=JobsHealthOperator.GREATER_THAN,
                value=3600
            ),
            JobsHealthRule(
                metric=JobsHealthMetric.STREAMING_BACKLOG_BYTES,
                op=JobsHealthOperator.GREATER_THAN,
                value=1073741824
            )
        ]
    ),
    tasks=[...]
)

Health Metrics

Operators

---

Timeout Configuration

Job-Level Timeout

resources:
  jobs:
    timeout_job:
      name: "Job with Timeout"
      timeout_seconds: 7200  # 2 hours max run time
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Task-Level Timeout

tasks:
  - task_key: long_running
    timeout_seconds: 3600  # 1 hour max for this task
    notebook_task:
      notebook_path: ../src/long_running.py

Python SDK

from databricks.sdk.service.jobs import Task, NotebookTask

Task(
    task_key="long_running",
    timeout_seconds=3600,
    notebook_task=NotebookTask(
        notebook_path="/Workspace/long_running"
    )
)

Timeout Behavior

• Value 0 = no timeout (default)
• When timeout exceeds, task/job is cancelled
• Partial results may be lost
• Triggers on_failure notifications

---

Retry Configuration

Task Retry Settings

tasks:
  - task_key: flaky_task
    max_retries: 3
    min_retry_interval_millis: 30000  # 30 seconds between retries
    retry_on_timeout: true
    notebook_task:
      notebook_path: ../src/flaky_task.py

Python SDK

from databricks.sdk.service.jobs import Task, NotebookTask

Task(
    task_key="flaky_task",
    max_retries=3,
    min_retry_interval_millis=30000,
    retry_on_timeout=True,
    notebook_task=NotebookTask(
        notebook_path="/Workspace/flaky_task"
    )
)

Retry Parameters

Retry Behavior

• Retries only apply to task failures
• Each retry is a new task attempt
• Retry count resets for each job run
• Dependencies wait for retries to complete

---

Run Queue Settings

Control concurrent run behavior.

Maximum Concurrent Runs

resources:
  jobs:
    concurrent_job:
      name: "Concurrent Job"
      max_concurrent_runs: 5  # Allow up to 5 simultaneous runs
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Queue Settings

resources:
  jobs:
    queued_job:
      name: "Queued Job"
      max_concurrent_runs: 1
      queue:
        enabled: true  # Queue additional runs instead of skipping
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import QueueSettings

w = WorkspaceClient()

job = w.jobs.create(
    name="Queued Job",
    max_concurrent_runs=1,
    queue=QueueSettings(enabled=True),
    tasks=[...]
)

Behavior Options

---

Notification Settings

Fine-tune notification behavior.

Job-Level Settings

resources:
  jobs:
    notification_settings_job:
      name: "Job with Notification Settings"
      notification_settings:
        no_alert_for_skipped_runs: true
        no_alert_for_canceled_runs: true
      email_notifications:
        on_failure:
          - "team@example.com"
      tasks:
        - task_key: main
          notebook_task:
            notebook_path: ../src/main.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import JobNotificationSettings

w = WorkspaceClient()

job = w.jobs.create(
    name="Job with Notification Settings",
    notification_settings=JobNotificationSettings(
        no_alert_for_skipped_runs=True,
        no_alert_for_canceled_runs=True
    ),
    tasks=[...]
)

Settings

---

Complete Monitoring Example

resources:
  jobs:
    fully_monitored:
      name: "[${bundle.target}] Fully Monitored ETL"

      # Timeout and retries
      timeout_seconds: 14400  # 4 hours max
      max_concurrent_runs: 1
      queue:
        enabled: true

      # Health monitoring
      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 7200  # Warn if > 2 hours

      # Email notifications
      email_notifications:
        on_start:
          - "team@example.com"
        on_success:
          - "team@example.com"
        on_failure:
          - "oncall@example.com"
          - "team@example.com"
        on_duration_warning_threshold_exceeded:
          - "oncall@example.com"
        no_alert_for_skipped_runs: true

      # Webhook notifications
      webhook_notifications:
        on_failure:
          - id: "pagerduty-destination-uuid"
        on_duration_warning_threshold_exceeded:
          - id: "slack-alerts-uuid"

      # Notification settings
      notification_settings:
        no_alert_for_canceled_runs: true

      tasks:
        - task_key: extract
          max_retries: 2
          min_retry_interval_millis: 60000
          timeout_seconds: 3600
          notebook_task:
            notebook_path: ../src/extract.py

        - task_key: transform
          depends_on:
            - task_key: extract
          max_retries: 1
          timeout_seconds: 3600
          notebook_task:
            notebook_path: ../src/transform.py

        - task_key: load
          depends_on:
            - task_key: transform
          timeout_seconds: 1800
          # Critical task - specific notifications
          email_notifications:
            on_failure:
              - "data-team-lead@example.com"
          notebook_task:
            notebook_path: ../src/load.py

Task Types Reference

Contents

• Notebook Task
• Spark Python Task
• Python Wheel Task
• SQL Task
• dbt Task
• Pipeline Task
• Spark JAR Task
• Run Job Task
• For Each Task

---

Notebook Task

Run Databricks notebooks. Most common task type.

Python SDK

from databricks.sdk.service.jobs import Task, NotebookTask, Source

Task(
    task_key="run_notebook",
    notebook_task=NotebookTask(
        notebook_path="/Workspace/Shared/etl_notebook",
        source=Source.WORKSPACE,
        base_parameters={
            "env": "prod",
            "date": "2024-01-15"
        }
    )
)

DABs YAML

tasks:
  - task_key: run_notebook
    notebook_task:
      notebook_path: ../src/notebooks/etl_notebook.py
      source: WORKSPACE
      base_parameters:
        env: "{{job.parameters.env}}"
        date: "{{job.parameters.date}}"

CLI JSON

{
  "task_key": "run_notebook",
  "notebook_task": {
    "notebook_path": "/Workspace/Shared/etl_notebook",
    "source": "WORKSPACE",
    "base_parameters": {
      "env": "prod",
      "date": "2024-01-15"
    }
  }
}

Parameters

Access Parameters in Notebook

# Get parameter with default
env = dbutils.widgets.get("env")

# Or define widget first
dbutils.widgets.text("env", "dev")
env = dbutils.widgets.get("env")

---

Spark Python Task

Run Python files directly on Spark cluster.

Python SDK

from databricks.sdk.service.jobs import Task, SparkPythonTask

Task(
    task_key="run_python",
    spark_python_task=SparkPythonTask(
        python_file="/Workspace/Shared/scripts/process.py",
        parameters=["--env", "prod", "--date", "2024-01-15"]
    )
)

DABs YAML

tasks:
  - task_key: run_python
    spark_python_task:
      python_file: ../src/scripts/process.py
      parameters:
        - "--env"
        - "prod"
        - "--date"
        - "2024-01-15"

CLI JSON

{
  "task_key": "run_python",
  "spark_python_task": {
    "python_file": "/Workspace/Shared/scripts/process.py",
    "parameters": ["--env", "prod", "--date", "2024-01-15"]
  }
}

Parameters

---

Python Wheel Task

Run Python packages distributed as wheels.

Python SDK

from databricks.sdk.service.jobs import Task, PythonWheelTask

Task(
    task_key="run_wheel",
    python_wheel_task=PythonWheelTask(
        package_name="my_package",
        entry_point="main",
        parameters=["--env", "prod"]
    ),
    libraries=[
        {"whl": "/Volumes/catalog/schema/libs/my_package-1.0.0-py3-none-any.whl"}
    ]
)

DABs YAML

tasks:
  - task_key: run_wheel
    python_wheel_task:
      package_name: my_package
      entry_point: main
      parameters:
        - "--env"
        - "prod"
    libraries:
      - whl: /Volumes/catalog/schema/libs/my_package-1.0.0-py3-none-any.whl

Parameters

Entry Point Configuration

In your package's setup.py or pyproject.toml:

# setup.py
entry_points={
    'console_scripts': [
        'main=my_package.main:run',
    ],
}

---

SQL Task

Run SQL queries, files, or refresh dashboards/alerts.

Run SQL Query

tasks:
  - task_key: run_query
    sql_task:
      query:
        query_id: "abc123-def456"  # Existing query ID
      warehouse_id: "1234567890abcdef"

Run SQL File

tasks:
  - task_key: run_sql_file
    sql_task:
      file:
        path: ../src/sql/transform.sql
        source: WORKSPACE
      warehouse_id: "1234567890abcdef"

Refresh Dashboard

tasks:
  - task_key: refresh_dashboard
    sql_task:
      dashboard:
        dashboard_id: "dashboard-uuid"
      warehouse_id: "1234567890abcdef"

Refresh Alert

tasks:
  - task_key: refresh_alert
    sql_task:
      alert:
        alert_id: "alert-uuid"
      warehouse_id: "1234567890abcdef"

Python SDK

from databricks.sdk.service.jobs import Task, SqlTask, SqlTaskFile

Task(
    task_key="run_sql",
    sql_task=SqlTask(
        warehouse_id="1234567890abcdef",
        file=SqlTaskFile(
            path="/Workspace/Shared/queries/transform.sql",
            source=Source.WORKSPACE
        )
    )
)

Parameters

---

dbt Task

Run dbt projects with Databricks.

DABs YAML

tasks:
  - task_key: run_dbt
    dbt_task:
      project_directory: ../src/dbt_project
      commands:
        - "dbt deps"
        - "dbt seed"
        - "dbt run --select tag:daily"
        - "dbt test"
      warehouse_id: "1234567890abcdef"
      catalog: "main"
      schema: "analytics"

Python SDK

from databricks.sdk.service.jobs import Task, DbtTask

Task(
    task_key="run_dbt",
    dbt_task=DbtTask(
        project_directory="/Workspace/Shared/dbt_project",
        commands=["dbt deps", "dbt run", "dbt test"],
        warehouse_id="1234567890abcdef",
        catalog="main",
        schema="analytics"
    )
)

Parameters

---

Pipeline Task

Trigger DLT or Spark Declarative Pipelines.

DABs YAML

tasks:
  - task_key: run_pipeline
    pipeline_task:
      pipeline_id: "pipeline-uuid-here"
      full_refresh: false

With Pipeline Resource Reference (DABs)

resources:
  pipelines:
    my_pipeline:
      name: "My Data Pipeline"
      # ... pipeline config

  jobs:
    my_job:
      name: "Orchestrate Pipeline"
      tasks:
        - task_key: run_pipeline
          pipeline_task:
            pipeline_id: ${resources.pipelines.my_pipeline.id}

Python SDK

from databricks.sdk.service.jobs import Task, PipelineTask

Task(
    task_key="run_pipeline",
    pipeline_task=PipelineTask(
        pipeline_id="pipeline-uuid-here",
        full_refresh=False
    )
)

Parameters

---

Spark JAR Task

Run Scala/Java JAR files on Spark.

DABs YAML

tasks:
  - task_key: run_jar
    spark_jar_task:
      main_class_name: "com.example.Main"
      parameters:
        - "--env"
        - "prod"
    libraries:
      - jar: /Volumes/catalog/schema/libs/my-app.jar

Python SDK

from databricks.sdk.service.jobs import Task, SparkJarTask

Task(
    task_key="run_jar",
    spark_jar_task=SparkJarTask(
        main_class_name="com.example.Main",
        parameters=["--env", "prod"]
    ),
    libraries=[
        {"jar": "/Volumes/catalog/schema/libs/my-app.jar"}
    ]
)

Parameters

---

Run Job Task

Trigger another job as a task (job chaining).

DABs YAML

tasks:
  - task_key: trigger_downstream
    run_job_task:
      job_id: 12345
      job_parameters:
        source_table: "catalog.schema.table"

With Job Resource Reference (DABs)

resources:
  jobs:
    upstream_job:
      name: "Upstream Job"
      tasks:
        - task_key: process
          notebook_task:
            notebook_path: ../src/process.py

    downstream_job:
      name: "Downstream Job"
      tasks:
        - task_key: trigger_upstream
          run_job_task:
            job_id: ${resources.jobs.upstream_job.id}

Python SDK

from databricks.sdk.service.jobs import Task, RunJobTask

Task(
    task_key="trigger_downstream",
    run_job_task=RunJobTask(
        job_id=12345,
        job_parameters={"source_table": "catalog.schema.table"}
    )
)

Parameters

---

For Each Task

Loop over a collection and run a nested task for each item.

DABs YAML - Static Inputs

tasks:
  - task_key: process_regions
    for_each_task:
      inputs: '["us-east", "us-west", "eu-west"]'
      task:
        task_key: process_region
        notebook_task:
          notebook_path: ../src/process_region.py
          base_parameters:
            region: "{{input}}"

DABs YAML - Dynamic Inputs from Previous Task

tasks:
  - task_key: generate_list
    notebook_task:
      notebook_path: ../src/generate_countries.py

  - task_key: process_countries
    depends_on:
      - task_key: generate_list
    for_each_task:
      inputs: "{{tasks.generate_list.values.countries}}"
      task:
        task_key: process_country
        notebook_task:
          notebook_path: ../src/process_country.py
          base_parameters:
            country: "{{input}}"

Generate Dynamic Inputs

In the generating notebook, return values using task values:

# generate_countries.py notebook
countries = ["USA", "UK", "Germany", "France"]

# Set task value for downstream for_each_task
dbutils.jobs.taskValues.set(key="countries", value=countries)

Python SDK

from databricks.sdk.service.jobs import Task, ForEachTask, NotebookTask

Task(
    task_key="process_regions",
    for_each_task=ForEachTask(
        inputs='["us-east", "us-west", "eu-west"]',
        task=Task(
            task_key="process_region",
            notebook_task=NotebookTask(
                notebook_path="/Workspace/process_region",
                base_parameters={"region": "{{input}}"}
            )
        )
    )
)

Parameters

Access Current Item

Inside the nested task, access the current item:

• In parameters: {{input}}
• In notebook: Use the parameter passed via base_parameters

---

Task Libraries

Add libraries to tasks for dependencies.

DABs YAML

tasks:
  - task_key: with_libraries
    notebook_task:
      notebook_path: ../src/notebook.py
    libraries:
      - pypi:
          package: pandas==2.0.0
      - pypi:
          package: scikit-learn
      - whl: /Volumes/catalog/schema/libs/custom-1.0.0-py3-none-any.whl
      - jar: /Volumes/catalog/schema/libs/custom.jar
      - maven:
          coordinates: "org.apache.spark:spark-avro_2.12:3.5.0"

Library Types

---

Environments

Define reusable Python environments for serverless tasks with custom pip dependencies.

IMPORTANT: The client field is required in the environment spec. It specifies the

base serverless environment version. Use "4" as the value. Without it, the API returns:

"Either base environment or version must be provided for environment".

DABs YAML

environments:
  - environment_key: ml_env
    spec:
      client: "4"
      dependencies:
        - pandas==2.0.0
        - scikit-learn==1.3.0
        - mlflow

tasks:
  - task_key: ml_task
    environment_key: ml_env
    notebook_task:
      notebook_path: ../src/train_model.py

CLI JSON

{
  "environments": [
    {
      "environment_key": "ml_env",
      "spec": {
        "client": "4",
        "dependencies": ["pandas==2.0.0", "scikit-learn==1.3.0"]
      }
    }
  ]
}

Python SDK

from databricks.sdk.service.jobs import JobEnvironment
from databricks.sdk.service.compute import Environment

environments = [
    JobEnvironment(
        environment_key="ml_env",
        spec=Environment(
            client="4",
            dependencies=["pandas==2.0.0", "scikit-learn==1.3.0"]
        )
    )
]

Parameters

Triggers and Schedules Reference

Contents

• Cron Schedule
• Periodic Trigger
• File Arrival Trigger
• Table Update Trigger
• Continuous Jobs
• Manual Runs

---

Cron Schedule

Run jobs on a cron-based schedule.

DABs YAML

resources:
  jobs:
    daily_etl:
      name: "Daily ETL"
      schedule:
        quartz_cron_expression: "0 0 8 * * ?"  # Daily at 8 AM
        timezone_id: "America/New_York"
        pause_status: UNPAUSED
      tasks:
        - task_key: etl
          notebook_task:
            notebook_path: ../src/etl.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import CronSchedule, PauseStatus

w = WorkspaceClient()

job = w.jobs.create(
    name="Daily ETL",
    schedule=CronSchedule(
        quartz_cron_expression="0 0 8 * * ?",
        timezone_id="America/New_York",
        pause_status=PauseStatus.UNPAUSED
    ),
    tasks=[...]
)

CLI JSON

{
  "name": "Daily ETL",
  "schedule": {
    "quartz_cron_expression": "0 0 8 * * ?",
    "timezone_id": "America/New_York",
    "pause_status": "UNPAUSED"
  },
  "tasks": [...]
}

Cron Expression Reference

Format: seconds minutes hours day-of-month month day-of-week

Common Timezones

---

Periodic Trigger

Run jobs at fixed intervals (simpler than cron).

DABs YAML

resources:
  jobs:
    hourly_sync:
      name: "Hourly Sync"
      trigger:
        pause_status: UNPAUSED
        periodic:
          interval: 1
          unit: HOURS
      tasks:
        - task_key: sync
          notebook_task:
            notebook_path: ../src/sync.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import TriggerSettings, Periodic, PeriodicTriggerConfigurationTimeUnit, PauseStatus

w = WorkspaceClient()

job = w.jobs.create(
    name="Hourly Sync",
    trigger=TriggerSettings(
        pause_status=PauseStatus.UNPAUSED,
        periodic=Periodic(
            interval=1,
            unit=PeriodicTriggerConfigurationTimeUnit.HOURS
        )
    ),
    tasks=[...]
)

Interval Units

Examples

# Every 30 minutes (not supported - use cron)
# Minimum periodic interval is 1 hour

# Every 4 hours
trigger:
  periodic:
    interval: 4
    unit: HOURS

# Every 2 days
trigger:
  periodic:
    interval: 2
    unit: DAYS

# Weekly
trigger:
  periodic:
    interval: 1
    unit: WEEKS

---

File Arrival Trigger

Run jobs when new files arrive in cloud storage.

DABs YAML

resources:
  jobs:
    process_uploads:
      name: "Process Uploads"
      trigger:
        pause_status: UNPAUSED
        file_arrival:
          url: "s3://my-bucket/uploads/"
          min_time_between_triggers_seconds: 60
          wait_after_last_change_seconds: 30
      tasks:
        - task_key: process
          notebook_task:
            notebook_path: ../src/process_files.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import TriggerSettings, FileArrivalTriggerConfiguration, PauseStatus

w = WorkspaceClient()

job = w.jobs.create(
    name="Process Uploads",
    trigger=TriggerSettings(
        pause_status=PauseStatus.UNPAUSED,
        file_arrival=FileArrivalTriggerConfiguration(
            url="s3://my-bucket/uploads/",
            min_time_between_triggers_seconds=60,
            wait_after_last_change_seconds=30
        )
    ),
    tasks=[...]
)

Parameters

Supported URL Formats

Access File Information in Notebook

# The trigger provides file information via task context
import json

# Get trigger info from job context
trigger_info = dbutils.jobs.taskValues.get(
    taskKey="__trigger_info__",
    key="file_arrival",
    debugValue={}
)

# Contains: url, files (list of new files)
print(f"New files: {trigger_info.get('files', [])}")

---

Table Update Trigger

Run jobs when Unity Catalog tables are updated.

DABs YAML

resources:
  jobs:
    process_updates:
      name: "Process Table Updates"
      trigger:
        pause_status: UNPAUSED
        table_update:
          table_names:
            - "catalog.schema.source_table"
            - "catalog.schema.other_table"
          condition: ANY_UPDATED
          min_time_between_triggers_seconds: 300
          wait_after_last_change_seconds: 60
      tasks:
        - task_key: process
          notebook_task:
            notebook_path: ../src/process_changes.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import (
    TriggerSettings,
    TableUpdateTriggerConfiguration,
    Condition,
    PauseStatus
)

w = WorkspaceClient()

job = w.jobs.create(
    name="Process Table Updates",
    trigger=TriggerSettings(
        pause_status=PauseStatus.UNPAUSED,
        table_update=TableUpdateTriggerConfiguration(
            table_names=["catalog.schema.source_table"],
            condition=Condition.ANY_UPDATED,
            min_time_between_triggers_seconds=300,
            wait_after_last_change_seconds=60
        )
    ),
    tasks=[...]
)

Parameters

Requirements

• Tables must be in Unity Catalog
• Job identity needs SELECT permission on monitored tables
• Works with Delta tables (managed and external)

---

Continuous Jobs

Always-running jobs that automatically restart.

DABs YAML

resources:
  jobs:
    streaming_job:
      name: "Streaming Processor"
      continuous:
        pause_status: UNPAUSED
      tasks:
        - task_key: stream
          notebook_task:
            notebook_path: ../src/streaming_processor.py

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Continuous, PauseStatus

w = WorkspaceClient()

job = w.jobs.create(
    name="Streaming Processor",
    continuous=Continuous(
        pause_status=PauseStatus.UNPAUSED
    ),
    tasks=[...]
)

Continuous Job Behavior

• Job runs immediately when created/unpaused
• Automatically restarts after completion or failure
• Maintains one active run at a time
• Use pause_status: PAUSED to stop

Control Continuous Jobs

# Pause continuous job
w.jobs.update(
    job_id=12345,
    new_settings=JobSettings(
        continuous=Continuous(pause_status=PauseStatus.PAUSED)
    )
)

# Resume continuous job
w.jobs.update(
    job_id=12345,
    new_settings=JobSettings(
        continuous=Continuous(pause_status=PauseStatus.UNPAUSED)
    )
)

---

Manual Runs

Run jobs on-demand without automatic triggers.

No Trigger Configuration

Simply omit schedule, trigger, and continuous:

resources:
  jobs:
    manual_job:
      name: "Manual Job"
      # No schedule/trigger = manual only
      tasks:
        - task_key: run
          notebook_task:
            notebook_path: ../src/manual_task.py

Trigger Manual Run

Python SDK:

# Run with default parameters
run = w.jobs.run_now(job_id=12345)

# Run with custom parameters
run = w.jobs.run_now(
    job_id=12345,
    job_parameters={"env": "prod", "date": "2024-01-15"}
)

# Wait for completion
run_result = w.jobs.run_now_and_wait(job_id=12345)

CLI:

# Run job
databricks jobs run-now 12345

# Run with parameters (must use --json with job_id inside)
databricks jobs run-now --json '{"job_id": 12345, "job_parameters": {"env": "prod"}}'

DABs:

databricks bundle run my_job_resource_key

---

Combining Triggers

A job can have multiple trigger types (evaluated independently):

resources:
  jobs:
    multi_trigger:
      name: "Multi-Trigger Job"
      # Cron schedule
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "UTC"
        pause_status: UNPAUSED
      # Also trigger on file arrival
      trigger:
        pause_status: UNPAUSED
        file_arrival:
          url: "s3://my-bucket/urgent/"
      tasks:
        - task_key: process
          notebook_task:
            notebook_path: ../src/process.py

Trigger Priority

When multiple triggers fire simultaneously:

• Job queues runs if max_concurrent_runs > 1
• Otherwise, subsequent triggers are skipped while a run is active

max_concurrent_runs: 1  # Only one run at a time (default)

---

Pause and Resume

Pause Scheduled Job

schedule:
  quartz_cron_expression: "0 0 8 * * ?"
  timezone_id: "UTC"
  pause_status: PAUSED  # Job won't run on schedule

Pause via SDK

from databricks.sdk.service.jobs import JobSettings, CronSchedule, PauseStatus

w.jobs.update(
    job_id=12345,
    new_settings=JobSettings(
        schedule=CronSchedule(
            quartz_cron_expression="0 0 8 * * ?",
            timezone_id="UTC",
            pause_status=PauseStatus.PAUSED
        )
    )
)

Pause via CLI

databricks jobs update 12345 --json '{
  "new_settings": {
    "schedule": {
      "pause_status": "PAUSED"
    }
  }
}'