BI Future Blog: 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 :

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

BI Future Blog

zondag 8 november 2026

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