title

Lakeflow Jobs — Part 2

type

topic

Lakeflow Jobs — Part 2

This part covers triggers and scheduling, compute configuration, notifications, job parameters, error handling, monitoring, common issues, and exam tips for Lakeflow Jobs (Lakeflow Jobs).

For job components, configuration, task dependencies, task values, and for-each loops, see Part 1.

Triggers and Scheduling

Databricks Job trigger schedule configuration showing cron, continuous, and file-arrival options.

Cron Schedule

schedule:
  quartz_cron_expression: "0 0 6 * * ?"  # Daily at 6 AM
  timezone_id: "America/New_York"
  pause_status: UNPAUSED

Common Cron Expressions

Expression	Description
`0 0 6 * * ?`	Daily at 6:00 AM
`0 0 * * * ?`	Every hour
`0 /15 * * ?`	Every 15 minutes
`0 0 6 ? * MON-FRI`	Weekdays at 6:00 AM
`0 0 6 1 * ?`	First of month at 6:00 AM
`0 0 6 ? * SUN`	Every Sunday at 6:00 AM

File Arrival Trigger

trigger:
  file_arrival:
    url: "s3://bucket/landing/data/"
    min_time_between_triggers_seconds: 60
    wait_after_last_change_seconds: 30

Continuous Trigger

trigger:
  periodic:
    interval: 1
    unit: HOURS

Compute Configuration

Job Clusters vs All-Purpose

Aspect	Job Cluster	All-Purpose Cluster
Lifecycle	Created/destroyed per job	Always running
Cost	Pay per job	Pay for uptime
Startup	Cold start delay	Immediate
Best for	Production jobs	Development

Job Cluster Configuration

job_clusters:
  - job_cluster_key: standard_etl
    new_cluster:
      spark_version: "14.3.x-scala2.12"
      node_type_id: "Standard_DS3_v2"
      num_workers: 2
      spark_conf:
        spark.databricks.delta.preview.enabled: "true"
        spark.sql.shuffle.partitions: "200"
      custom_tags:
        purpose: etl
        environment: ${var.environment}

  - job_cluster_key: large_processing
    new_cluster:
      spark_version: "14.3.x-scala2.12"
      node_type_id: "Standard_DS4_v2"
      autoscale:
        min_workers: 2
        max_workers: 10
      spark_conf:
        spark.databricks.adaptive.autoOptimizeShuffle.enabled: "true"

Serverless Compute

tasks:
  - task_key: serverless_task
    notebook_task:
      notebook_path: ../notebooks/process.py
    # Uses serverless compute (no cluster config)
    environment_key: default  # Or custom environment

Notifications

Email Notifications

email_notifications:
  on_start:
    - team@company.com
  on_success:
    - team@company.com
  on_failure:
    - team@company.com
    - oncall@company.com
  on_duration_warning_threshold_exceeded:
    - team@company.com
  no_alert_for_skipped_runs: true

Webhook Notifications

webhook_notifications:
  on_start:
    - id: ${var.slack_webhook_id}
  on_success:
    - id: ${var.slack_webhook_id}
  on_failure:
    - id: ${var.pagerduty_webhook_id}
    - id: ${var.slack_webhook_id}

Duration Warning

# Warn if job exceeds expected duration

health:
  rules:
    - metric: RUN_DURATION_SECONDS
      op: GREATER_THAN
      value: 1800  # 30 minutes

Job Parameters

Widget Parameters

# In notebook

dbutils.widgets.text("date", "2024-01-15")
dbutils.widgets.dropdown("environment", "dev", ["dev", "staging", "prod"])
dbutils.widgets.text("source_path", "/mnt/landing/")

# Get values

date = dbutils.widgets.get("date")
env = dbutils.widgets.get("environment")
source = dbutils.widgets.get("source_path")

Job-Level Parameters

parameters:
  - name: date
    default: "2024-01-15"
  - name: environment
    default: "dev"

tasks:
  - task_key: process
    notebook_task:
      notebook_path: ../notebooks/process.py
      base_parameters:
        date: "{{job.parameters.date}}"
        environment: "{{job.parameters.environment}}"

Dynamic Parameters

# Run job with parameters via CLI

databricks jobs run-now --job-id 12345 \
    --notebook-params '{"date": "2024-01-20", "mode": "full"}'

# Via API

curl -X POST "$WORKSPACE_URL/api/2.1/jobs/run-now" \
    -H "Authorization: Bearer $TOKEN" \
    -d '{"job_id": 12345, "notebook_params": {"date": "2024-01-20"}}'

Error Handling

Retry Configuration

tasks:
  - task_key: potentially_flaky_task
    notebook_task:
      notebook_path: ../notebooks/external_api.py
    # Retry on failure
    max_retries: 3
    min_retry_interval_millis: 60000  # 1 minute
    retry_on_timeout: true
    timeout_seconds: 600

Failure Handling Task

tasks:
  - task_key: main_processing
    notebook_task:
      notebook_path: ../notebooks/main.py

  - task_key: handle_failure
    depends_on:
      - task_key: main_processing
    run_if: AT_LEAST_ONE_FAILED
    notebook_task:
      notebook_path: ../notebooks/failure_handler.py

Monitoring and Logging

Job run matrix showing historical success and failure status across multiple executions.

Job Run API

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Get job runs

runs = w.jobs.list_runs(job_id=12345, limit=10)
for run in runs:
    print(f"Run {run.run_id}: {run.state.result_state}")

# Get specific run details

run = w.jobs.get_run(run_id=67890)
print(f"Duration: {run.run_duration / 1000} seconds")
print(f"State: {run.state.life_cycle_state}")

System Tables for Job Monitoring

-- Query job run history
SELECT
    job_id,
    run_id,
    run_name,
    result_state,
    ROUND(run_duration / 1000, 0) AS duration_seconds,
    start_time,
    end_time
FROM system.lakeflow.job_run_timeline
WHERE job_id = 12345
ORDER BY start_time DESC
LIMIT 100;

-- Task-level metrics
SELECT
    job_id,
    run_id,
    task_key,
    result_state,
    ROUND(execution_duration / 1000, 0) AS duration_seconds
FROM system.lakeflow.job_task_run_timeline
WHERE job_id = 12345
ORDER BY start_time DESC;

Use Cases

Cross-Workspace Data Promotion: Configuring a multi-task job that sequentially triggers an ingestion pipeline, executes a suite of PyTest validation code, and conditionally alerts stakeholders via Slack if any task fails using the AT_LEAST_ONE_FAILED condition.
Scheduled Feature Engineering: Running a parameterized machine learning feature extraction notebook on a cron schedule using an isolated, cost-effective job cluster to process recent interactions before model retraining occurs.

Common Issues & Errors

Task Timeout

Scenario: Task exceeds timeout and is killed.

Fix: Adjust timeout or optimize task:

tasks:
  - task_key: long_running
    notebook_task:
      notebook_path: ../notebooks/heavy_processing.py
    timeout_seconds: 7200  # 2 hours

Cluster Start Failure

Scenario: Job cluster fails to start.

Fix: Check cluster configuration and quotas:

job_clusters:
  - job_cluster_key: fallback_cluster
    new_cluster:
      spark_version: "14.3.x-scala2.12"
      # Use smaller instance if quota issues
      node_type_id: "Standard_DS2_v2"
      num_workers: 1

Task Value Not Found

Scenario: Downstream task can't read task value.

Fix: Verify task key and key name:

# Upstream task - verify this sets the value

dbutils.jobs.taskValues.set(key="result", value="success")

# Downstream task - use exact task key

value = dbutils.jobs.taskValues.get(
    taskKey="upstream_task_key",  # Must match exactly
    key="result",
    default="unknown"
)

Circular Dependencies

Scenario: Job fails validation due to circular deps.

Fix: Review and fix dependency graph:

# Wrong - circular

tasks:
  - task_key: a
    depends_on: [{task_key: b}]
  - task_key: b
    depends_on: [{task_key: a}]

# Correct - acyclic

tasks:
  - task_key: a
  - task_key: b
    depends_on: [{task_key: a}]

Exam Tips

Task types - Notebook, Pipeline, SQL, Python, JAR, If/Else, For Each
Dependencies - depends_on defines execution order; tasks with no dependencies run in parallel
Task values - Inter-task communication with dbutils.jobs.taskValues.set/get
Run conditions - ALL_SUCCESS, AT_LEAST_ONE_FAILED, ALL_DONE
Job clusters - Created and destroyed per run; cost-effective for production
Cron syntax - Quartz format (7 fields including seconds); ? means "no specific value"
File arrival trigger - Starts job when new files are detected at a path
Retry config - max_retries and min_retry_interval_millis per task
For each - Parallel processing of array items from upstream task values
Notifications - Email and webhook notifications on start, success, and failure

Key Takeaways

Task execution order: Tasks with no depends_on entries run in parallel; depends_on enforces sequential execution and the dependency graph must be a DAG (no cycles allowed).
Run conditions: ALL_SUCCESS (default) runs only if all upstream tasks succeed; AT_LEAST_ONE_FAILED creates a failure-handler task; ALL_DONE runs regardless of upstream outcome.
Task values for inter-task data: Use dbutils.jobs.taskValues.set(key, value) in upstream tasks and dbutils.jobs.taskValues.get(taskKey, key) in downstream tasks to share computed values within a run.
Job clusters are cost-effective: Job clusters are created fresh per run and terminated on completion, costing ~60% less than keeping an All-Purpose cluster running 24/7.
Quartz cron format: Databricks uses 7-field Quartz cron (includes seconds field); ? means "no specific value" and is used in day-of-month or day-of-week when the other is specified.
File arrival trigger: The file_arrival trigger starts a job when new files are detected at a cloud storage path, eliminating the need for polling-based scheduling.
Retry configuration: Set max_retries and min_retry_interval_millis per task to handle transient failures; retry_on_timeout: true retries tasks that exceed timeout_seconds.
system.lakeflow tables: Query system.lakeflow.job_run_timeline and job_task_run_timeline for programmatic job run history, duration metrics, and result states.

Official Documentation

← Previous: Lakeflow Jobs — Part 1 | ↑ Back to Lakeflow Pipelines

Navigation Menu

Lakeflow Jobs — Part 2

Triggers and Scheduling

Cron Schedule

Common Cron Expressions

File Arrival Trigger

Continuous Trigger

Compute Configuration

Job Clusters vs All-Purpose

Job Cluster Configuration

Serverless Compute

Notifications

Email Notifications

Webhook Notifications

Duration Warning

Job Parameters

Widget Parameters

Job-Level Parameters

Dynamic Parameters

Error Handling

Retry Configuration

Failure Handling Task

Monitoring and Logging

Job Run API

System Tables for Job Monitoring

Use Cases

Common Issues & Errors

Task Timeout

Cluster Start Failure

Task Value Not Found

Circular Dependencies

Exam Tips

Key Takeaways

Related Topics

Official Documentation

FilesExpand file tree

07-lakeflow-jobs-part2.md

Latest commit

History

07-lakeflow-jobs-part2.md

File metadata and controls

Lakeflow Jobs — Part 2

Triggers and Scheduling

Cron Schedule

Common Cron Expressions

File Arrival Trigger

Continuous Trigger

Compute Configuration

Job Clusters vs All-Purpose

Job Cluster Configuration

Serverless Compute

Notifications

Email Notifications

Webhook Notifications

Duration Warning

Job Parameters

Widget Parameters

Job-Level Parameters

Dynamic Parameters

Error Handling

Retry Configuration

Failure Handling Task

Monitoring and Logging

Job Run API

System Tables for Job Monitoring

Use Cases

Common Issues & Errors

Task Timeout

Cluster Start Failure

Task Value Not Found

Circular Dependencies

Exam Tips

Key Takeaways

Related Topics

Official Documentation