update conditional execution example

Author: zanitarahimi

knowledge_base/pydabs_job_conditional_execution/README.md

@@ -6,13 +6,13 @@ The Lakeflow Job consists of following tasks:
 1. Checks data quality and calculates bad records
 2. Evaluates if bad records exceed a threshold (100 records)
 3. Routes to different processing paths based on the condition:
-   - If bad records > 100: runs `handle_bad_data` task
-   - If bad records ≤ 100: runs `continue_pipeline` task
+   - If bad records > 100: runs `fix_path` task
+   - If bad records ≤ 100: runs `skip_path` task
 
 * `src/`: Notebook source code for this project.
   * `src/check_quality.ipynb`: Checks data quality and outputs bad record count
-  * `src/process_bad_data.ipynb`: Handles cases with high bad record count
-  * `src/process_good_data.ipynb`: Continues normal pipeline for good data
+  * `src/fix_path.ipynb`: Handles cases with high bad record count
+  * `src/skip_path.ipynb`: Continues normal pipeline for good data
 * `resources/`:  Resource configurations (jobs, pipelines, etc.)
   * `resources/conditional_execution.py`: PyDABs job definition with conditional tasks
 
@@ -55,10 +55,10 @@ with this project. It's also possible to interact with it directly using the CLI
 
     This deploys everything that's defined for this project.
     For example, this project will deploy a job called
-    `[dev yourname] pydabs_job_conditional_execution` to your workspace.
+    `[dev yourname] conditional_execution_example` to your workspace.
     You can find that resource by opening your workspace and clicking on **Jobs & Pipelines**.
 
 3. To run the job, use the "run" command:
    ```
-   $ databricks bundle run pydabs_job_conditional_execution
+   $ databricks bundle run conditional_execution_example
    ```

knowledge_base/pydabs_job_conditional_execution/resources/conditional_execution.py

@@ -1,36 +1,43 @@
 from databricks.bundles.jobs import (
-    ConditionTask,
     Job,
-    NotebookTask,
     Task,
+    NotebookTask,
+    ConditionTask,
+    ConditionTaskOp,
     TaskDependency,
 )
 
-pydabs_job_conditional_execution = Job(
-    name="pydabs_job_conditional_execution",
-    tasks=[
-        Task(
-            task_key="check_data_quality",
-            notebook_task=NotebookTask(notebook_path="src/check_quality.ipynb"),
-        ),
-        Task(
-            task_key="evaluate_quality",
-            condition_task=ConditionTask(
-                left="{{tasks.check_data_quality.values.bad_records}}",
-                op="GREATER_THAN",
-                right="100",
-            ),
-            depends_on=[TaskDependency(task_key="check_data_quality")],
-        ),
-        Task(
-            task_key="handle_bad_data",
-            notebook_task=NotebookTask(notebook_path="src/process_bad_data.ipynb"),
-            depends_on=[TaskDependency(task_key="evaluate_quality", outcome="true")],
-        ),
-        Task(
-            task_key="continue_pipeline",
-            notebook_task=NotebookTask(notebook_path="src/process_good_data.ipynb"),
-            depends_on=[TaskDependency(task_key="evaluate_quality", outcome="false")],
-        ),
-    ],
+# 1) Producer task: runs a notebook and emits a task value
+check_quality = Task(
+    task_key="check_quality",
+    notebook_task=NotebookTask(notebook_path="src/branch/check_quality.ipynb"),
+)
+
+# 2) Branch task: evaluates an expression using an upstream task value
+branch = Task(
+    task_key="branch",
+    condition_task=ConditionTask(
+        left="{{tasks.check_quality.values.bad_records}}",
+        op=ConditionTaskOp.GREATER_THAN,
+        right="100",
+    ),
+    depends_on=[TaskDependency(task_key="check_quality")],
+)
+
+# 3) Downstream tasks: gated on the condition outcome
+fix_path = Task(
+    task_key="fix_path",
+    notebook_task=NotebookTask(notebook_path="src/branch/fix_path.ipynb"),
+    depends_on=[TaskDependency(task_key="branch", outcome="true")],
+)
+
+skip_path = Task(
+    task_key="skip_path",
+    notebook_task=NotebookTask(notebook_path="src/branch/skip_path.ipynb"),
+    depends_on=[TaskDependency(task_key="branch", outcome="false")],
+)
+
+job = Job(
+    name="conditional_execution_example",
+    tasks=[check_quality, branch, fix_path, skip_path],
 )