diff --git a/knowledge_base/job_orchestration_edge_cases/.gitignore b/knowledge_base/job_orchestration_edge_cases/.gitignore new file mode 100644 index 0000000..15bcc6d --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/.gitignore @@ -0,0 +1 @@ +.databricks diff --git a/knowledge_base/job_orchestration_edge_cases/README.md b/knowledge_base/job_orchestration_edge_cases/README.md new file mode 100644 index 0000000..07d4c87 --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/README.md @@ -0,0 +1,49 @@ +# Job orchestration edge cases + +This Declarative Automation Bundles example defines a serverless job with `run_if` conditions, task retries, run queueing, and `for_each` tasks. + +One `for_each` iteration fails on purpose, so the run ends in a controlled partial failure. + +For configuration details, see the [run if documentation](https://docs.databricks.com/aws/en/jobs/run-if). + +## Prerequisites + +- Databricks CLI v1.20.0 or above +- Configure a Databricks CLI profile for the target workspace. +- Use a workspace that supports serverless jobs environment version `6`. + +## Usage + +1. Deploy the job. The deploy stops if the job has an active run. + + ```bash + databricks bundle deploy --fail-on-active-runs + ``` + +2. Run the job. + + ```bash + databricks bundle run orchestration_edge_cases + ``` + +3. Examine the run result. The command fails because the run result is `FAILED`. The tasks have these results: + + | Task | `run_if` | Result | + | --- | --- | --- | + | `process_items` | `ALL_SUCCESS` | `FAILED`: item `beta` fails twice, with one retry | + | `process_no_items` | `ALL_SUCCESS` | `SUCCESS`: zero iterations | + | `publish` | `ALL_SUCCESS` | `UPSTREAM_FAILED` | + | `cleanup` | `ALL_DONE` | `SUCCESS` | + | `alert_on_failure` | `AT_LEAST_ONE_FAILED` | `SUCCESS` | + +4. Show the result of each task. Replace `RUN_ID` with the run ID from the run URL. + + ```bash + databricks jobs get-run RUN_ID + ``` + +5. Remove the deployed resources. + + ```bash + databricks bundle destroy + ``` diff --git a/knowledge_base/job_orchestration_edge_cases/databricks.yml b/knowledge_base/job_orchestration_edge_cases/databricks.yml new file mode 100644 index 0000000..bfaeb12 --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/databricks.yml @@ -0,0 +1,17 @@ +bundle: + name: job_orchestration_edge_cases + +include: + - resources/*.job.yml + +targets: + dev: + default: true + mode: development + workspace: + root_path: /Workspace/bundle-examples/${bundle.name}/${bundle.target}/${workspace.current_user.short_name} + + prod: + mode: production + workspace: + root_path: /Workspace/bundle-examples/${bundle.name}/${bundle.target} diff --git a/knowledge_base/job_orchestration_edge_cases/resources/orchestration_edge_cases.job.yml b/knowledge_base/job_orchestration_edge_cases/resources/orchestration_edge_cases.job.yml new file mode 100644 index 0000000..f748e28 --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/resources/orchestration_edge_cases.job.yml @@ -0,0 +1,85 @@ +resources: + jobs: + orchestration_edge_cases: + name: Job orchestration edge cases + description: Demonstrates run_if, retries, queueing, and for_each with a failed iteration. + + # Allow one active run. Queue extra runs instead of skipping them. + max_concurrent_runs: 1 + queue: + enabled: true + + tasks: + # Each input is a JSON object. Access its fields with {{input.}}. + - task_key: process_items + for_each_task: + inputs: '[{"name": "alpha", "fail": false}, {"name": "beta", "fail": true}, {"name": "gamma", "fail": false}]' + concurrency: 3 + task: + task_key: process_item + environment_key: default + spark_python_task: + python_file: ../src/process_item.py + parameters: + - --name + - "{{input.name}}" + - --fail + - "{{input.fail}}" + # Retry a failed iteration once after 10 seconds. + timeout_seconds: 300 + max_retries: 1 + min_retry_interval_millis: 10000 + retry_on_timeout: true + + # An empty input list runs zero iterations and succeeds. + - task_key: process_no_items + for_each_task: + inputs: "[]" + task: + task_key: process_no_item + environment_key: default + spark_python_task: + python_file: ../src/process_item.py + parameters: + - --name + - "{{input.name}}" + + # Runs only if all dependencies succeed (the default ALL_SUCCESS). + - task_key: publish + depends_on: + - task_key: process_items + - task_key: process_no_items + environment_key: default + spark_python_task: + python_file: ../src/step.py + parameters: + - publish + + # Runs after all dependencies finish, whatever their result. + - task_key: cleanup + depends_on: + - task_key: process_items + - task_key: process_no_items + run_if: ALL_DONE + environment_key: default + spark_python_task: + python_file: ../src/step.py + parameters: + - cleanup + + # Runs only if at least one dependency fails. + - task_key: alert_on_failure + depends_on: + - task_key: process_items + - task_key: process_no_items + run_if: AT_LEAST_ONE_FAILED + environment_key: default + spark_python_task: + python_file: ../src/step.py + parameters: + - alert_on_failure + + environments: + - environment_key: default + spec: + environment_version: "6" diff --git a/knowledge_base/job_orchestration_edge_cases/src/process_item.py b/knowledge_base/job_orchestration_edge_cases/src/process_item.py new file mode 100644 index 0000000..537abed --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/src/process_item.py @@ -0,0 +1,12 @@ +import argparse + + +parser = argparse.ArgumentParser() +parser.add_argument("--name", required=True) +parser.add_argument("--fail", choices=("true", "false"), default="false") +args = parser.parse_args() + +if args.fail == "true": + raise RuntimeError(f"Item {args.name} failed on purpose") + +print(f"Processed item {args.name}") diff --git a/knowledge_base/job_orchestration_edge_cases/src/step.py b/knowledge_base/job_orchestration_edge_cases/src/step.py new file mode 100644 index 0000000..c6f8a81 --- /dev/null +++ b/knowledge_base/job_orchestration_edge_cases/src/step.py @@ -0,0 +1,4 @@ +import sys + + +print(f"Running step {sys.argv[1]}")