update for each example

Author: zanitarahimi

knowledge_base/pydabs_job_with_for_each/README.md

@@ -3,7 +3,8 @@
 This example demonstrates a simple Databricks job that uses a foreach task.
 
 * `src/`: Python source code for this project.
-  * `src/pydabs_job_with_for_each/`: Shared Python code that can be used by jobs and pipelines.
+    * `foreach/generate_items.ipynb`: A notebook which returns a list of items to be used for task generation.
+    * `foreach/process_item.ipynb`: A notebook which will process an item.
 * `resources/`:  Resource configurations (jobs, pipelines, etc.)
 
 
@@ -45,7 +46,7 @@ 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, the default template would deploy a job called
-    `[dev yourname] pydabs_airflow_job` to your workspace.
+    `[dev yourname] for_each_task_example` to your workspace.
     You can find that resource by opening your workpace and clicking on **Jobs & Pipelines**.
 
 3. Similarly, to deploy a production copy, type:

knowledge_base/pydabs_job_with_for_each/resources/for_each_simple.py

@@ -1,38 +1,35 @@
-from databricks.bundles.jobs import Job, Task, NotebookTask, ForEachTask, TaskDependency
-
-extract = Task(
-    task_key="extract",
-    notebook_task=NotebookTask(notebook_path="src/notebook_extract.py"),
+from databricks.bundles.jobs import (
+    Job,
+    Task,
+    NotebookTask,
+    ForEachTask,
+    TaskDependency,
+    JobEnvironment,
+    Environment,
 )
-process_item_iteration = Task(
-    task_key="process_item_iteration",
-    notebook_task=NotebookTask(
-        notebook_path="src/notebook_process_item.py",
-        base_parameters={
-            "index": "{{input}}",
-        },
-    ),
+
+generate_items = Task(
+    task_key="generate_items",
+    notebook_task=NotebookTask(notebook_path="src/foreach/generate_items.ipynb"),
 )
+
 process_item = Task(
     task_key="process_item",
-    depends_on=[TaskDependency(task_key="extract")],
     for_each_task=ForEachTask(
-        inputs="{{tasks.extract.values.indexes}}",
-        task=process_item_iteration,
+        inputs="{{tasks.generate_items.values.items}}",
+        task=Task(
+            task_key="process_item_iteration",
+            notebook_task=NotebookTask(
+                notebook_path="src/foreach/process_item.ipynb",
+                base_parameters={"item": "{{input}}"},
+            ),
+        ),
         concurrency=10,
     ),
+    depends_on=[TaskDependency(task_key="generate_items")],
 )
 
-for_each_example = Job(
-    name="for_each_example",
-    tasks=[
-        extract,
-        process_item,
-    ],
-    parameters=[
-        {
-            "name": "lookup_file_name",
-            "default": "/Volumes/main/for_each_example/hotchpotch/my_file.json",
-        },
-    ],
-)
+for_each_task_example = Job(
+    name="for_each_task_example",
+    tasks=[generate_items, process_item],
+)

knowledge_base/pydabs_job_with_for_each/src/foreach/generate_items.ipynb

@@ -0,0 +1,24 @@
+{
+  "cells": [
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "id": "e4c15846",
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "from databricks.sdk.runtime import dbutils\n",
+        "\n",
+        "items = [1, 2, 3]\n",
+        "dbutils.jobs.taskValues.set(key=\"items\", value=items)"
+      ]
+    }
+  ],
+  "metadata": {
+    "language_info": {
+      "name": "python"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 5
+}

knowledge_base/pydabs_job_with_for_each/src/foreach/process_item.ipynb

@@ -0,0 +1,25 @@
+{
+  "cells": [
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "# Runs once per item in the for-each. Do not call dbutils.jobs.taskValues.set() here.\n",
+        "from databricks.sdk.runtime import dbutils\n",
+        "\n",
+        "# Current iteration value passed from the for-each task (base_parameters: item = {{input}})\n",
+        "current_item = dbutils.widgets.get(\"item\")\n",
+        "print(f\"Processing item: {current_item}\")"
+      ]
+    }
+  ],
+  "metadata": {
+    "language_info": {
+      "name": "python"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 5
+}