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