Databricks series : Execute the job in Databricks (Part VII)

Introduction

This blogpost is the seventh in a series of blogposts about Databricks. In these series I showed and will show several features of Databricks with easy to understand examples. In this blogpost we'll describe a more configurable ingestion framework.

Previous blogposts : 

• Databricks series : Starting a new python project in Visual Studio Code (Part I) 
• Databricks series : setting up a connection to Databricks from local IDE (Part II)
• Databricks series : Setting up a Databricks project in Visual Studio Code (part III)
• Databricks series : Deployment from Visual Studio Code (Part IV)
• Databricks series : Execute the job in Databricks (Part V)
• Databricks series : Execute the job in Databricks (Part VI)

Currently, we have a Python script that reads a CSV file and writes it to a Delta table. Now we will take the next step in building a data engine.

We support three loading strategies:

• Overwrite. The existing table is replaced by the new dataset. Suitable for full snapshots.
• Append. New records are added to the existing table. Suitable for transactional data.
• Merge. Existing records are updated, and new records are added. Suitable for changes to customer data.

For this, we will modify only one file: src/transfer/processor.py.

Extend the loading strategies

from delta.tables import DeltaTable

def transfer_data(spark, config):

    source = config["source"]
    target = config["target"]
    file_format = config["format"]
    mode = config["mode"]

    df = (
        spark.read
        .format(file_format)
        .option("header", "true")
        .option("inferSchema", "true")
        .load(source)
    )

    if mode == "overwrite":

        df.write.format("delta") \
            .mode("overwrite") \
            .saveAsTable(target)

    elif mode == "append":

        df.write.format("delta") \
            .mode("append") \
            .saveAsTable(target)

    elif mode == "merge":

        merge_key = config["merge_key"]

        if spark.catalog.tableExists(target):

            delta_table = DeltaTable.forName(spark, target)

            (
                delta_table.alias("target")
                .merge(
                    df.alias("source"),
                    f"target.{merge_key} = source.{merge_key}"
                )
                .whenMatchedUpdateAll()
                .whenNotMatchedInsertAll()
                .execute()
            )

        else:

            df.write.format("delta") \
                .mode("overwrite") \
                .saveAsTable(target)

    else:
        raise ValueError(f"Unsupported mode: {mode}")

    print(f"Successfully processed {df.count()} records")

Note: For a Merge, we add a 'merge_key' to the configuration. We must also ensure that the source data does not contain duplicate keys; otherwise, the Merge may fail.

Open `config/entities.json` and modify it as follows:

{
  "customers": {
    "source": "/Volumes/main/demo/raw/customers.csv",
    "target": "main.demo.customers",
    "format": "csv",
    "mode": "overwrite"
  },
  "customers_update": {
    "source": "/Volumes/main/demo/raw/customers_update.csv",
    "target": "main.demo.customers",
    "format": "csv",
    "mode": "merge",
    "merge_key": "customer_id"
  },
  "transactions": {
    "source": "/Volumes/main/demo/raw/transactions.csv",
    "target": "main.demo.transactions",
    "format": "csv",
    "mode": "append"
  }
}

The configuration now determines how the data is processed. You no longer need to modify the Python code when you want to use a different loading strategy for an entity.

Testing with modified customer data

To properly demonstrate the Merge functionality, we will create a second customers file containing modified data. We name the file : customers_update.csv

customer_id,first_name,last_name,country
1001,Jan,Jansen,Netherlands
1002,Pieter,Peeters,Belgium
1005,Sarah,Johnson,USA

We use two files: customers.csv

customer_id First name Last name    Country

1001         Jan                Jansen             Netherlands

1002         Piet                  Peeters            Belgium

1003         Anna          Smith              United Kingdom

1004         Marie          Dubois             France

And we have customers_update.csv :

customer_id First name Last name        Country

1001         Jan                 Jansen              Netherlands

1002         Pieter         Peeters               Belgium

1005         Sarah         Johnson            USA

After a MERGE, we expect the following:

• 1001 remains unchanged.
• 1002 is updated from Piet to Pieter.
• 1003 remains.
• 1004 remains.
• 1005 is added.

Upload the file to Databricks

Deploying to Databricks

Now we can deploy the project. Open the terminal in Visual Studio Code and navigate to the root of your project.  Execute:

databricks bundle validate -t dev

Resulting in : 

And if everything is right : 

databricks bundle deploy -t dev

Next, you can start the job:

databricks bundle run transfer_job -t dev --params entity=customers

Check in Databricks SQL 

SELECT *
FROM main.demo.customers
ORDER BY customer_id

This will give the following results :

Second run: customers_update

Now for the interesting part.  We run the same Job again, but with a different parameter:

databricks bundle run transfer_job -t dev --params entity=customers_update

Check in Databricks SQL :

SELECT *
FROM main.demo.customers
ORDER BY customer_id

Resulting in :

And so it succeeds quite well. Piet has been changed to Pieter and we have a new record. 

Now to be complete, lets test the append.

databricks bundle run transfer_job -t dev --params entity=transactions 

and again : 

databricks bundle run transfer_job -t dev --params entity=transactions 

And we test the table we have duplicate records in the table

Thats it!

Hennie

Databricks series : Execute the job in Databricks (Part VI)

Introduction

This blogspost is the sixth in a series of blogposts about databricks. In these series I showed and will show several features of Databricks with easy to understand examples. In this blogpost we'll describe a more configurable ingestion framework. we'll replace the parameters in the job with a configrable file.

Previous blogposts : 

• Databricks series : Starting a new python project in Visual Studio Code (Part I) 
• Databricks series : setting up a connection to Databricks from local IDE (Part II)
• Databricks series : Setting up a Databricks project in Visual Studio Code (part III)
• Databricks series : Deployment from Visual Studio Code (Part IV)
• Databricks series : Execute the job in Databricks (Part V)

Currently, we have a Python script that reads a CSV file and writes it to a Delta table.

We want to extend this so that we no longer need to hardcode the source, target table, and loading strategy for each dataset within the Python code.

Instead, we will create a single configuration file that defines what needs to be done. The major advantage is that if you add a new entity tomorrow, you essentially only need to expand the configuration.

The new project structure

First, lets open the existing project, 'databricks-transfer', in Visual Studio Code. We are going to add a new directory called 'config'. We will add a file entities.json in there.

The project should now look like this:

databricks-transfer/
│
├── databricks.yml
│
├── resources/
│   └── transfer_job.yml
│
├── src/
│   ├── __init__.py
│   └── main.py
│
├── config/
│   └── entities.json
│
├── data/
│   ├── customers.csv
│   └── transactions.csv
│
├── requirements.txt
└── README.md

This file contains the configuration for all the entities we want to process.

{
  "customers": {
    "source": "/Volumes/main/demo/raw/customers.csv",
    "target": "main.demo.customers",
    "format": "csv",
    "mode": "overwrite"
  },
  "transactions": {
    "source": "/Volumes/main/demo/raw/transactions.csv",
    "target": "main.demo.transactions",
    "format": "csv",
    "mode": "append"
  },
  "products": {
    "source": "/Volumes/main/demo/raw/products.json",
    "target": "main.demo.products",
    "format": "json",
    "mode": "overwrite"
  }
}

Let’s take a look at what we are doing here.

• customers. This is the name of the entity
• source. The location of the source file
• target. The target table in Unity Catalog
• format. The file format
• mode. The loading strategy

We have now defined three datasets. Note: 'products.json' is included as an example for a future expansion. You do not need to create that file yet. Why are we doing this? Suppose you want to add a fourth entity later, for example, 'suppliers'. In that case, you simply add an extra configuration block. You don’t need to write a new Python function.

Create a separate Python module for the configuration

Now, we will actually read the configuration. Open the 'src' directory. Create a new directory inside it:

src/
├── __init__.py
├── main.py
└── transfer/
    ├── __init__.py
    └── config_loader.py

Create the file src/transfer/config_loader.py. Add the following to it:

import json


def load_entity_config(config_path, entity):

    with open(config_path, "r") as file:
        config = json.load(file)

    if entity not in config:
        raise ValueError(
            f"Unknown entity: {entity}"
        )

    return config[entity]

What does this function do?

1. Opens the JSON configuration file.

2. Reads the content into a Python dictionary.

3. Looks up the requested entity.

4. Returns the corresponding configuration.

For example, when the 'customers' job is passed, we get:

{
    "source": "/Volumes/main/demo/raw/customers.csv",
    "target": "main.demo.customers",
    "format": "csv",
    "mode": "overwrite"
}

Create a separate Python module for data transfer

We will move the actual data transfer logic out of main.py. Create a new file: 'src/transfer/processor.py'. Add the following to it:

from pyspark.sql import SparkSession


def transfer_data(spark, config):

    source = config["source"]
    target = config["target"]
    file_format = config["format"]
    mode = config["mode"]

    print(f"Reading source: {source}")
    print(f"Target table: {target}")
    print(f"Format: {file_format}")
    print(f"Mode: {mode}")

    # Read source file
    df = (
        spark.read
        .format(file_format)
        .option("header", "true")
        .load(source)
    )

    # Write to Delta table
    (
        df.write
        .format("delta")
        .mode(mode)
        .saveAsTable(target)
    )

    print(f"Processed {df.count()} records")

We have now divided the responsibilities:

• config_loader.py: determines what needs to be done.
• processor.py: carries out the data transfer.

This is an important step in developing maintainable Python applications.

Modify main.py

Now, let’s bring everything together. Open src/main.py. Replace the existing content with:

import argparse
from pathlib import Path

from pyspark.sql import SparkSession

from transfer.config_loader import load_entity_config
from transfer.processor import transfer_data


def main():

    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--entity",
        required=True
    )

    args = parser.parse_args()

    spark = SparkSession.builder.getOrCreate()

    # Configuration file location
    config_path = (
        Path(__file__).resolve().parents[1]
        / "config"
        / "entities.json"
    )

    # Load entity configuration
    config = load_entity_config(
        config_path,
        args.entity
    )

    # Execute data transfer
    transfer_data(spark, config)


if __name__ == "__main__":
    main()

What has changed? Previously, we provided three parameters:

entity
source_path
target_table

But now, we only provide:

entity

The rest is retrieved from entities.json. That is exactly what we want to achieve!

Ensure that DAB also deploys the configuration file

This is an important step that you must not overlook. Your Python code is located in `src`, while our JSON file is in `config`. We need to ensure that both directories are included in the deployment. Open `databricks.yml`. Modify its content:

bundle:
  name: databricks-transfer

include:
  - resources/*.yml

sync:
  include:
    - src/**
    - config/**
    - requirements.txt

targets:
  dev:
    default: true
    mode: development

Now, both the Python files and the JSON configuration file are synchronized.

Adjust the Databricks Job

Since we are now passing only the entity, we also need to update our Job configuration. Open: resources/transfer_job.yml and replace the contents with:

resources:
  jobs:
    transfer_job:

      name: python-data-transfer

      parameters:
        - name: entity
          default: customers

      tasks:
        - task_key: transfer_data

          spark_python_task:
            python_file: ../src/main.py

            parameters:
              - --entity
              - "{{job.parameters.entity}}"

          environment_key: default

      environments:
        - environment_key: default
          spec:
            environment_version: "2"

Our Job now has only one parameter: `entity`. When running it, you can choose from the following options: 'customers', 'transactions,  'products'. The Python code and the Job itself do not need to be modified for this.

Step 8 – Deploying to Databricks

Now we can deploy the project. Open the terminal in Visual Studio Code and navigate to the root of your project.  Execute:

databricks bundle validate -t dev

Resulting in : 

And if everything is right : 

databricks bundle deploy -t dev

Next, you can start the job:

> databricks bundle run transfer_job -t dev

Run URL: https://dbc-xxxxxx-d21b.cloud.databricks.com/jobs/846415909479637/runs/750647492568451?o=1134000616576115

2026-11-08 13:31:06 "[dev xxxxx] python-data-transfer" RUNNING
2026-11-08 13:31:47 "[dev xxxxx] python-data-transfer" TERMINATED SUCCESS
Output:
=======
Task transfer_data:
Processing entity: customers

=======
Task transfer_data:
Processing entity: customers
Successfully processed 4 records

The default value is customers.

Do you want to perform transactions? You can override the parameter when starting:

databricks bundle run transfer_job -t dev \
  --params entity=transactions

The parameter is then passed to main.py.

Important: ensure that the configuration files are actually available at the deployed path. In the example above, we use a path relative to main.py; therefore, the directory structure must be preserved during a DAB deployment.

The tables are now created : 

That's it!

Hennie

Databricks series : Execute the job in Databricks (Part V)

Introduction

This blogspost is the fifth in a series of blogposts. This blogpost describes the execution of the job in Databricks. In order to do that we need to do one thing and that is we need to place the files at the raw folder. Next is to execute job and as the final part we examine the results.

Previous blogposts : 

• Databricks series : Starting a new python project in Visual Studio Code (Part I) 
• Databricks series : setting up a connection to Databricks from local IDE (Part II)
• Databricks series : Setting up a Databricks project in Visual Studio Code (part III)
• Databricks series : Deployment from Visual Studio Code (Part IV)

Create a volume in databricks

First we need to create a folder structure at the location :  /Volumes/main/demo/raw/

Open a notebook in Databricks and execute the following code :

%sql
CREATE CATALOG IF NOT EXISTS main;

CREATE SCHEMA IF NOT EXISTS main.demo;

CREATE VOLUME IF NOT EXISTS main.demo.raw;

This means : 

main                       ← Catalog

 └── demo            ← Schema

      └── raw          ← Volume

Upload the files to the location :

And here are the files stored at the locations. Everything is ready to execute the job.

Uitvoeren van de job

You can upload the files from our local data folder for this purpose.

Subsequently, we can run the same job twice.

entity = customers

source_path = /Volumes/main/demo/raw/customers.csv

target_table = main.demo.customers

entity = transactions

source_path  = /Volumes/main/demo/raw/transactions.csv

target_table = main.demo.transactions

The result should be: 

Job         Delta-table                 Records

Run 1 main.demo.customers 4

Run 2 main.demo.transactions 4

So, we have used one Python script and one Job for two different data streams.

Open de job and press on Run Now

The job is running now

Here you find information which code is executed in the job.

More information is available in the Job execution window

And in the unity catalog we see the following information.

Now execute the job with other parameters!

Enter the parameters and execute again.

And here information about the second run of the job

And this results in two tables with customer and transaction information.

Hennie

Databricks series : Deployment from Visual Studio Code (Part IV)

Introduction 

I've written a couple of blogposts about a small project to create a Databricks project. This blogpost is the fourth in the series. Now the fun part: we don’t have to manually create the Job in the Databricks interface. We’ll use the Databricks CLI for this!

Previous blogposts : 

• Databricks series : Starting a new python project in Visual Studio Code (Part I) 
• Databricks series : setting up a connection to Databricks from local IDE (Part II)
• Databricks series : Setting up a Databricks project in Visual Studio Code (part III)

Deploy the code

Open a terminal window in Visual Studio Code and authenticate with databricks auth login

 > databricks auth login --host https://dbc-xxxxxx-d21b.cloud.databricks.com/
 --profile DBprofile

Validate the bundle

Next step is validate the code with databricks bundle validate

> databricks bundle validate

results in : 

Name: DatabricksProject
Target: dev
Workspace:
  Host: https://dbc-xxxxxx-d21b.cloud.databricks.com
  User: xxxxxx@gmail.com
  Path: /Workspace/Users/xxxxx@gmail.com/.bundle/DatabricksProject/dev

Deploy the code

And deploy the code with databricks bundle deploy

> databricks bundle deploy

results in : 

Uploading bundle files to /Workspace/Users/xxxxr@gmail.com/.bundle/DatabricksProject/dev/files...
Created jobs.transfer_job
Files: 1 uploaded, 0 deleted
Resources: 1 created, 0 changed, 0 deleted, 0 unchanged

Databricks now deploys:

• The Python code.
• The job configuration.
• The required project files.

Where is everything in Databricks?

In your Databricks Workspace, go to: Workspace → Users → your user → .bundle

There, you will typically find a structure such as:

Job-configuration

The Job itself is not necessarily found as a YAML file within the Workspace.

Go to: Workflows → Jobs & Pipelines

Search for: python-data-transfer

There, you can view:

• The Job configuration.
• The configured parameters.
• The Python Script Task.
• The execution history.
• The logs and results.

The YAML configuration remains part of your local DAB project in Visual Studio Code. Databricks uses that configuration to create or update the Job.

So that's it. The code is deployed to Databricks with DAB

Hennie

Databricks series : Setting up a Databricks project in Visual Studio Code (part III)

Introduction

In this blogpost I'll describe the Visual Studio Project in more detail. This blogpost is one in a series of blogposts. Here are the previous blogposts : 

• Databricks series : Starting a new python project in Visual Studio Code (Part I) 
• Databricks series : setting up a connection to Databricks from local IDE (Part II)

Project structure

Lets create a project structure where we can store our code and also the data files. I've created the following structure in Visual studio Code :

databricks-transfer/
│
├── databricks.yml
├── resources/
│   └── transfer_job.yml
│
├── src/
│   ├── __init__.py
│   └── main.py
│
├── data/
│   ├── customers.csv
│   └── transactions.csv
│
├── requirements.txt
└── README.md

What is the function of the different folders?

Some test data

For this blogseries, I've created a couple of testfiles with help of ChatGPT :

customers.csv

transactions.csv

Python code

The application performs three tasks:

1. Receives parameters from the Databricks Job.

2. Reads the specified CSV file using Spark.

3. Writes the data to a Delta table.

The advantage is that the Python code is independent of the specific dataset being processed.

import argparse
from pyspark.sql import SparkSession


def main():

    parser = argparse.ArgumentParser()

    parser.add_argument("--entity", required=True)
    parser.add_argument("--source-path", required=True)
    parser.add_argument("--target-table", required=True)

    args = parser.parse_args()

    spark = SparkSession.builder.getOrCreate()

    print(f"Processing entity: {args.entity}")

    df = (
        spark.read
        .option("header", "true")
        .option("inferSchema", "true")
        .csv(args.source_path)
    )

    (
        df.write
        .format("delta")
        .mode("overwrite")
        .saveAsTable(args.target_table)
    )

    print(f"Successfully processed {df.count()} records")


if __name__ == "__main__":
    main()

DAB

DAB will start with databricks.yml file. This is the content of the databricks.yml:

bundle:
  name: DatabricksProject

include:
  - resources/*.yml

targets:
  dev:
    mode: development
    default: true
    workspace:
      host: https://dbc-xxxxxxx.cloud.databricks.com

This is an example of a serverless Python Job. The environment is linked to the task via environment_key. The Job has three parameters:

Create a transfer_job.yml in the resources folder :

resources:
  jobs:
    transfer_job:

      name: python-data-transfer

      parameters:
        - name: entity
          default: customers

        - name: source_path
          default: /Volumes/main/demo/raw/customers.csv

        - name: target_table
          default: main.demo.customers

      tasks:
        - task_key: transfer_data

          spark_python_task:
            python_file: ../src/main.py

            parameters:
              - --entity
              - "{{job.parameters.entity}}"
              - --source-path
              - "{{job.parameters.source_path}}"
              - --target-table
              - "{{job.parameters.target_table}}"

          environment_key: default

      environments:
        - environment_key: default
          spec:
            environment_version: "2"

The Job passes these parameters to main.py. Pay special attention to:"{{job.parameters.entity}}"

This is a dynamic reference to a Job parameter. Databricks populates this value when the Job is executed. This means we do not need to modify our Python code when we want to process a different dataset. It is also possible to perform an explicit sync action.

bundle:
  name: DatabricksProject

include:
  - resources/*.yml

sync:
  include:
    - src/**
    - requirements.txt
    
targets:
  dev:
    mode: development
    default: true
    workspace:
      host: https://dbc-xxxxxx-d21b.cloud.databricks.com

  ## Optionally, there could be 'staging' or 'prod' targets here.
  #
  # prod:
  #   workspace:
  #     host: https://dbc-xxxxxx-d21b.cloud.databricks.com

This allows you to specify which project files you want to synchronize. We deliberately exclude the `data/` directory: our CSV files are ultimately stored in a Unity Catalog Volume.

Hennie