In my last post, I mentioned my concept of a Pure Python UI library called Schorle. I’ve put some more effort into it, making the API a bit more Pythonic and simplifying it for rendering purposes.

Let’s see how a to-do list web app can be implemented with almost Pure Python (we still use a bit of CSS for styling, but it’s pretty much inevitable).

What’s new?

Recent changes I’ve made provide capabilities for quite interesting functionality:

1. Conditional rendering
2. Rendering of lists
3. Suspense and loading fallback

All these features play quite nicely with the example I’ve implemented. You can take a look at the full source code here, and play with it a bit by installing the schorle package.

Components

While cooking Schorle, I’ve taken some good practices from other UI libraries. One of the abstractions I like in particular is the abstraction of a Component from React. Practically, it allows to combine various HTML elements, as well as other components into logical blocks.

However, one thing that I don’t like about components in React is that you need to use JSX to add some logic (e.g. if-else or for each operation). Considering these ideas, I came up with the following API in Python:

class TodoView(Component):
    state: State
    classes: Classes = Classes("max-w-96 w-2/3 space-y-2 flex flex-col items-center")

    def render(self):
        with p(classes=Classes("text-xl")):
            _text = "No todos yet." if not self.state.todos else "Your todos:"
            text(_text)

        for todo in self.state.todos:
            with div(classes=Classes("flex flex-row items-center justify-between w-full")):
                with p():
                    text(todo)
                with Button(
                    on=On("click", partial(self.state.remove, todo)),
                    modifier="error",
                ):
                    text("Delete")

    def initialize(self):
        self.suspense = Suspense(on=self.state, fallback=Loading())
        self.bind(self.state)

Let’s stop here for a second to consider what’s happening.

In the render method of a component, the layout is described. Since it’s a valid Python code, any standard Pythonic construction can be used, for example, the if-else conditioning on the text, or a for loop to display all of the todos.

The state class provides access for read and write operations on the page state. Using the On class, it’s quite easy to define a server-side callback to the click trigger.

Indeed, the component needs to be re-rendered on any change of the state. To do so, thebind method is used. Also, since the operations on the state might take time, suspense is added to the component.

What’s happening in the background?

From the client side, prepared HTML will look like this:

<div class="max-w-96 w-2/3 space-y-2 flex flex-col items-center" id="sle-div-7be95dec" hx-swap-oob="morph">
    <p class="text-xl" hx-swap-oob="morph">Your todos:</p>
    <div class="flex flex-row items-center justify-between w-full" hx-swap-oob="morph">
        <p hx-swap-oob="morph">Buy milk</p>
        <button class="btn btn-error" hx-swap-oob="morph" ws-send="" hx-trigger="click" id="sle-059f8171">Delete</button>
    </div>
    <div class="flex flex-row items-center justify-between w-full" hx-swap-oob="morph">
        <p hx-swap-oob="morph">Do laundry</p>
        <button class="btn btn-error" hx-swap-oob="morph" ws-send="" hx-trigger="click" id="sle-a561de43">Delete</button>
    </div>
</div>

The generated IDs are used for client-server communication. For instance, deleting an item will cause the following message:

{
    "HEADERS": {
        "HX-Request": "true",
        "HX-Trigger": "sle-e164c325",
        "HX-Trigger-Name": null,
        "HX-Target": "sle-e164c325",
        "HX-Current-URL": "http://127.0.0.1:8000/",
        "HX-Trigger-Type": "click"
    }
}

At first, the server will respond with a loading view:

<div id="sle-div-ed017a60" hx-swap-oob="morph">
  <div id="sle-div-7f1e0849">
    <span class="loading loading-lg loading-infinity"></span>
  </div>
</div>

And then, as soon as the operation is finished, a proper response will come:

<div class="max-w-96 w-2/3 space-y-2 flex flex-col items-center" id="sle-div-ed017a60" hx-swap-oob="morph">
  <p class="text-xl" hx-swap-oob="morph">Your todos:</p>
  <div class="flex flex-row items-center justify-between w-full" hx-swap-oob="morph">
    <p hx-swap-oob="morph">Buy milk</p>
    <button class="btn btn-error" hx-swap-oob="morph" ws-send="" hx-trigger="click" id="sle-ff9a97ae">Delete</button>
  </div>
</div>

On the UI, all of this will look as follows:

State operations

Since I’ve mentioned state, let’s take a look at the implementation of it:

class State(ReactiveModel):
    current: str = ""
    todos: list[str] = Field(default_factory=lambda: ["Buy milk", "Do laundry"])

    @effector
    async def set_current(self, value):
        self.current = value

    @effector
    async def add_todo(self):
        await asyncio.sleep(random() * 2)  # simulate a slow backend
        self.todos.append(self.current)
        await self.set_current("")

    @effector
    async def remove(self, todo):
        await asyncio.sleep(random() * 2)  # simulate a slow backend
        self.todos.remove(todo)

ReactiveModel is a class provided by the schorle framework. To simplify the explanation, consider it as a standard Pydantic BaseModel, with some necessary additions to make @effector work.

The @effector decorator highlights that this method has effects on the UI. Calling it outside of the schorle context, without any binds won’t affect the logic of the methods inside.

Text input

Since we also need to match the input together with the “Add” button, we need some way to exchange the input information with the server. Fortunately, we can use the same state class to do so, and add another UI component to combine this all together:

class InputSection(Component):
    state: State

    def render(self):
        with div(classes=Classes("flex flex-row w-3/4 justify-center items-center space-x-4")):
            TextInput(
                value=self.state.current,
                placeholder="Add a todo",
                classes=Classes("input-bordered"),
                on=On("change", self.state.set_current),
            )
            with Button(on=On("click", self.state.add_todo), modifier="primary"):
                text("Add")

    def initialize(self):
        self.bind(self.state)

Since the render is triggered only when any effectors on the state are fired, it will always correctly display the value of the input. When the user clicks Add, the value saved into current is used to add the todo.

Summary

Altogether, it took only 98 lines of code to prepare quite a functional to-do list using the framework, which I find particularly good.

Take note that we don’t use any frontend code, there is no need to JSONify the inputs back and forth between the server and the client. In this sense, the Schorle framework closely follows the Hypermedia As The Engine Of Application State (HATEAOS) principles.

What I find the most interesting is that with such an architecture developers can immediately interop with the inputs on the server side. For instance, to add the todo information into a database, the only required thing is to add saving methods to the State class defined above.

Under the hood, schorle indeed stands on the shoulders of giants:

• FastAPI — web framework
• HTMX — client-side library for dynamic HTML
• Tailwind CSS — CSS framework
• DaisyUI — Component library for Tailwind CSS
• Pydantic — classes, and utilities for elements

I find a combination of FastAPI and HTMX very powerful, and the idea of the framework is to expose this power to the developers without having too much hassle with various low-level details.

CSS styling is almost inevitable for a properly extensible framework, and here Tailwind + DaisyUI seems like the best fit from the existing toolset.

If you’re interested in the concepts or ideas implemented in Schorle, I would be happy to hear your feedback. Try it out yourselves, and let me know if there are any issues or ideas on what could make it a better framework ☺️

Preface

I’ve built a small, and not-yet-polished asynchronous framework in Python for server-side UI development, called schorle . It lacks a lot of features of mature frameworks, but it has several things that are conceptually different from the existing solutions.

In this blog post, I would like to consider these differences, describe the concepts, and maybe spark some community interest. If after reading you’ll feel the urge to try out (or maybe even contribute), take a look at the repo with the source code.

Existing approaches in JS and Python

Modern web UIs inevitably use one of the following approaches at their core:

• CSR (Client-Side Rendering)
• SSR (Server-Side Rendering)

Examples of CSR frameworks are React, Vue, and Svelte. These are great technologies that deliver an uncompromised level of UI controls and convenience for the developers. However, from a backend developer point of view, they require not only knowing the basics of JS (or TS) but also quite a good understanding of how the browser works, loads the pages, etc.

These frameworks are reaching out to the server side, e.g. to request data or interact with the server. To perform such interactions, developers on the front-end side use browser APIs like fetch to retrieve and push data back to the server.

On the other hand, there is the SSR approach, when the server renders the whole page into an HTML payload and returns it as a response to the request from the browser.

Indeed all of these frameworks support SSR in some or another way, and there is a particularly interesting feature in the Next.JS framework called ISR.

Such cool features of the JS ecosystem show a valid path for Server-Side applications in other languages, and there are many ways to do this in Python (the language I’m particularly used to).

Here are some examples of how this could be done:

• Rendering HTML pages from Jinja templates (e.g. in Flask, Django, FastAPI)
• Using powerful UI frameworks such as Dash and Streamlit

However, I see three missing things in these approaches in Python:

• Async and incremental page updates (with a strong accent on async)
• Strong typing for the UI elements
• Simple multi-page routing, auth, and other server-side features
• Well-typed state management

With these ideas in mind, I made a small PoC project schorle — and here is my conceptual vision of what a good Server-Driven UI kit in Python might look like.

Concepts

When it comes to writing a UI application, I’ve considered 3 main things that need to be covered:

• Elements and layout of the pages
• Reacting to client-side events
• Reacting to server-side state changes

As per the list of “potential features” described above, I also would like to extend these definitions to:

• Well-typed elements and layouts, with the possibility of re-using them
• Asynchronous reactivity for client-side events
• Asynchronous reactivity for server-side state changes

Let’s consider these 3 features in a simple example — a UI with 2 buttons and a shared state between them:

As we can see, this app has all three concepts in place:

• We have a layout of elements
• Two of these elements are reactive to the client-side actions
• One element is a “receiver” that changes when the server-side state is changed.

Starting from the layout, let’s see how it’s managed in Schorle:

from __future__ import annotations

from schorle.app import Schorle
from schorle.effector import effector
from schorle.elements.button import Button
from schorle.elements.html import Div
from schorle.elements.page import Page, PageReference
from schorle.reactives.classes import Classes
from schorle.reactives.state import ReactiveModel
from schorle.reactives.text import Text

# snip

class DecrementButton(Button):
    text: Text = Text("Decrement")
    page: PageWithButton = PageReference()
    classes: Classes = Classes("btn-error")
    # snip


class Buttons(Div):
    classes: Classes = Classes("space-x-4 flex flex-row justify-center items-center")
    increment: Button = Button.factory(text=Text("Increment"), classes=Classes("btn-success"))
    decrement: DecrementButton = DecrementButton.factory()
    page: PageWithButton = PageReference()
    # snip


class CounterView(Div):
    # snip

class PageWithButton(Page):
    counter: Counter = Counter.factory()
    classes: Classes = Classes("space-y-4 flex flex-col justify-center items-center h-screen w-screen")
    buttons: Buttons = Buttons.factory()
    counter_view: CounterView = CounterView.factory()

In the framework, elements are represented as a small subclass around pydantic BaseModel class. This provides a convenient way to introduce new elements via subclassing them, as well as an interface to combine them into more complex structures (e.g. nested Div with two buttons as per the code above).

Elements have two special fields, namely text and classes . These fields provide a convenient way to change the text and classes on the element.

Now, let’s add some reactivity to these components:

class DecrementButton(Button):
    text: Text = Text("Decrement")
    page: PageWithButton = PageReference()
    classes: Classes = Classes("btn-error")

    async def _switch_off(self, counter: Counter):
        if counter.value <= 0:
            await self.classes.append("btn-disabled")
        else:
            await self.classes.remove("btn-disabled")

class Buttons(Div):
    classes: Classes = Classes("space-x-4 flex flex-row justify-center items-center")
    increment: Button = Button.factory(text=Text("Increment"), classes=Classes("btn-success"))
    decrement: DecrementButton = DecrementButton.factory()
    page: PageWithButton = PageReference()

    async def before_render(self):
        self.increment.add_callback("click", self.page.counter.increment)
        self.decrement.add_callback("click", self.page.counter.decrement)

class CounterView(Div):
    page: PageWithButton = PageReference()

    async def update(self, counter: Counter):
        await self.text.update(f"Clicked {counter.value} times")

With this piece of code, we add reactivity to the following UI elements:

• The DecrementButton will become disabled as soon as the counter is equal to or less than zero
• The counter view needs to be updated with the new value of the counter
• Both buttons shall be reactive and therefore we add callbacks to them.

This code is indeed not runnable, because we’re missing a class that represents the Counter itself. Let’s introduce it as follows:

class Counter(ReactiveModel):
    value: int = 0

    @effector
    async def increment(self):
        self.value += 1

    @effector
    async def decrement(self):
        self.value -= 1

This class seems quite strange at first glance. Why do functions that operate on the value have this strange decorator, and more importantly — why do they need to be async?

Indeed this is not a coincidence — this is where we introduce the feedback loop — when something changes on the server, we need to send updates back to the client UI.

We can think of such a server-side change as an effector (function which has effects), and the UI update in this case is an effect.

With this logical concept, we can introduce a way to connect the server-side actions, e.g. functional calls to the UI updates.

In schorle, it’s possible to subscribe other methods to the effectors as follows:

class DecrementButton(Button):
    text: Text = Text("Decrement")
    page: PageWithButton = PageReference()
    classes: Classes = Classes("btn-error")

    async def before_render(self):
        await self.page.counter.decrement.subscribe(self._switch_off, trigger=True)
        await self.page.counter.increment.subscribe(self._switch_off, trigger=True)

    async def _switch_off(self, counter: Counter):
        if counter.value <= 0:
            await self.classes.append("btn-disabled")
        else:
            await self.classes.remove("btn-disabled")

class CounterView(Div):
    page: PageWithButton = PageReference()

    async def update(self, counter: Counter):
        await self.text.update(f"Clicked {counter.value} times")

    async def before_render(self):
        await self.page.counter.increment.subscribe(self.update, trigger=True)
        await self.page.counter.decrement.subscribe(self.update, trigger=True)

In this example, we say that:

• Whenever an increment or decrement happens, we need to change the classes of the DecrementButton
• Whenever an increment or decrement happens, we need to update the text of the counter view.

Now we can put this logic together and have the following code:

from __future__ import annotations

from schorle.app import Schorle
from schorle.effector import effector
from schorle.elements.button import Button
from schorle.elements.html import Div
from schorle.elements.page import Page, PageReference
from schorle.reactives.classes import Classes
from schorle.reactives.state import ReactiveModel
from schorle.reactives.text import Text

app = Schorle()


class Counter(ReactiveModel):
    value: int = 0

    @effector
    async def increment(self):
        self.value += 1

    @effector
    async def decrement(self):
        self.value -= 1


class DecrementButton(Button):
    text: Text = Text("Decrement")
    page: PageWithButton = PageReference()
    classes: Classes = Classes("btn-error")

    async def before_render(self):
        await self.page.counter.decrement.subscribe(self._switch_off, trigger=True)
        await self.page.counter.increment.subscribe(self._switch_off, trigger=True)

    async def _switch_off(self, counter: Counter):
        if counter.value <= 0:
            await self.classes.append("btn-disabled")
        else:
            await self.classes.remove("btn-disabled")


class Buttons(Div):
    classes: Classes = Classes("space-x-4 flex flex-row justify-center items-center")
    increment: Button = Button.factory(text=Text("Increment"), classes=Classes("btn-success"))
    decrement: DecrementButton = DecrementButton.factory()
    page: PageWithButton = PageReference()

    async def before_render(self):
        self.increment.add_callback("click", self.page.counter.increment)
        self.decrement.add_callback("click", self.page.counter.decrement)


class CounterView(Div):
    page: PageWithButton = PageReference()

    async def update(self, counter: Counter):
        await self.text.update(f"Clicked {counter.value} times")

    async def before_render(self):
        await self.page.counter.increment.subscribe(self.update, trigger=True)
        await self.page.counter.decrement.subscribe(self.update, trigger=True)


class PageWithButton(Page):
    counter: Counter = Counter.factory()
    classes: Classes = Classes("space-y-4 flex flex-col justify-center items-center h-screen w-screen")
    buttons: Buttons = Buttons.factory()
    counter_view: CounterView = CounterView.factory()


@app.get("/")
def get_page():
    return PageWithButton()

I’m not entirely convinced that this is good enough (I assume some simplifications might still be applied), but I see some good start here:

• It follows the imperative paradigm of Python programming, explicitly defining what should be changed and when.
• Functions that have side effects are defined explicitly, and the side effects are managed explicitly, almost following the Zen of Python in the concepts of Explicit is better than implicit .
• Elements, effects, and functions are well-typed and the behavior is predictable.

Logically, the flow of events looks like this:

Under the hood, the following technologies are used:

1. HTMX on the client side to define triggers and communicate messages via WebSockets
2. Pydantic-based classes to represent elements and state
3. FastAPI as a server backend
4. TailwindCSS and DaisyUI for effortless styling

Summary

The project is currently in a PoC state, meaning that I’m looking for more feedback and ideas from the Python community on how such things could be done better. Share your thoughts in the comments, or maybe even try creating some apps with it, and let me know how that feels from a developer perspective.

Intelligent data application adoption is rapidly growing amongst various data platforms. And with this growing adoption, engineers are looking for efficient and flexible APIs to work with their data in a way they’re used to operating pipelines, but inside such data applications.

In this blog post, I’ll demonstrate how the upgraded Databricks Connect protocol (based on Spark Connect) enables developers to write concise and expressive data applications on top of the Databricks Lakehouse Platform with the Dash framework.

⛓️ The Tale of Protocols

There are several ways to perform operations with the data stored in the Databricks Lakehouse from the data application.

One of them is the one I’ve described in one of my previous blog posts — it’s a straightforward and well-known approach that uses DBSQL and the SQLAlchemy ORM mechanism.

However, this time I would like to highlight the (relatively) new functionality of Databricks, called Databricks Connect “V2”. Although for many of the Databricks users, this component may sound familiar, its new version provides a more flexible and faster approach within a thinner version of a client that can be potentially used from applications in various programming languages.

Why would a Data Engineer prefer using Databricks Connect over a standard ORM-like solution?

There are several reasons for that, and one of the most important for me is that most of the Data Engineers have very good knowledge of Apache Spark and its APIs, whilst SQLAlchemy is usually closer to the experience of Software Engineers.

Being a Data Engineer myself in the past, between the two APIs below I clearly know which one I would like to choose most:

👁️ Solution overview

The architecture for such an app is relatively simple — there are two main components, namely:

• the frontend app (written in the robust Dash framework in almost pure Python)
• the backend (which turns out to be a Databricks Cluster)

Connectivity and data exchange between the backend and frontend is achieved via DB Connect “V2” protocol and the databricks-connect Python package.

The image metadata (e.g. classes) is stored in a Delta Table, while the images themselves are stored in a cloud storage as files.

The main benefits of this architecture are the following:

• Images can be stored and retrieved natively, without a need to put them into tables as blobs or binaries
• The metadata of the image can be easily changed and modified due to Delta Lake UPDATE functionality
• We don’t need to use a specific ORM and can perform the backend operations by using Spark API methods itself
• Our frontend app can be fully written in Python, without a need to switch language contexts

💻 Setting up the development environment

Prerequisites for this step:

• [Local machine] JVM (I’m using JVM 11 in this example)
• [Local machine] Poetry
• [Local machine] Databricks profile configured via databricks-cli
• [Databricks] Databricks All-purpose Cluster with DBR 13.X or higher and Unity Catalog support

Let’s start by organizing the project and the dev environment. Our app is a poetry -based Python package, and we can easily kick off the project with all dependencies as follows:

# create project dir
mkdir db-connect-v2-image-classification

# move into project dir
cd db-connect-v2-image-classification

# initialize poetry
poetry init -n \
  --dependency="databricks-connect=13.*" \ 
  --dependency="hydra-core=1.3.*" \ # for configurations
  --dependency="databricks-sdk" \ # for cluster ops
  --dependency="dash" # frontend framework

# add dev dependencies
poetry add -G dev black ruff isort

# poetry references this file, let's keep it empty for this moment
touch README.md

# add package directory
mkdir db_connect_v2_image_classification
touch db_connect_v2_image_classification/__init__.py
# install project and it's deps
poetry install

With these steps, we have a fully prepared Python app skeleton. Let’s step into the project and add the first bit that is responsible for preparing the metadata table:

code . # open VS Code in the project directory

Add a .gitignore file with the following contents:

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

# Distribution / packaging
*.egg-info/
build
dist

# Unit test / coverage reports
.coverage
coverage.xml
junit/*
htmlcov/*

# Caches
.pytest_cache/

# VSCode
.vscode/

# Idea
.idea/
*.iml

# MacOS
.DS_Store

And then init the git project:

git init .

Hint — you can immediately create a repo in GitHub if you have GitHub CLI installed and if you’re using Bash-like solution:

gh repo create "${PWD##*/}" --private --source=. --remote=origin

Now we have a proper project skeleton, and it’s pretty convenient to work with it. However our app will manage several configurations — both the connection-related ones, as well as ETL and data related ones.

🪄 Adding configurations

One of the most convenient ways to handle configurations is to use the hydra-core library.

Let’s add the config-related logic:

# db_connect_v2_image_classification/configs.py

from dataclasses import dataclass, field
from typing import Optional

from databricks.sdk import WorkspaceClient


@dataclass
class ImageTableConfig:
    catalog: str = "main"
    database: str = "default"
    table: str = "db_connect_v2_example_images_metadata"

    # note - repr is used both inside print statements and when table name is provided to Spark APIs
    def __repr__(self) -> str:
        return f"{self.catalog}.{self.database}.{self.table}"


@dataclass
class AppConfig:
    cluster_name: str
    cluster_id: str = field(init=False)
    image_table: ImageTableConfig = field(default_factory=ImageTableConfig)
    profile: Optional[str] = "DEFAULT"
    debug: Optional[bool] = True

    # defines is we want to overwite the table, fail when it exists or append data into it
    table_saving_mode: Optional[str] = "error"

    def __post_init__(self):
        _w = WorkspaceClient(profile=self.profile)
        found = list(filter(lambda c: c.cluster_name == self.cluster_name, _w.clusters.list()))
        assert found is not None, f"Not found any clusters with name {self.cluster_name}"
        assert len(found) == 1, f"There are more than one cluster with name {self.cluster_name}"
        self.cluster_id = found[0].cluster_id

The code above introduces data classes and a logical structure for the application config. Since dataclasses have a convenient __post_init__ method, we can automatically resolve the provided cluster_name argument into a cluster_id by using the latest Databricks SDK for Python.

🌯 Wrapping the entrypoints

With the configuration structure defined, it’s also important to add a bit of a glue code so hydrawould work with the configurations in a way it would be convenient for our app:

# db_connect_v2_image_classification/main_wrapper.py

from typing import Callable

import hydra
from hydra.core.config_store import ConfigStore
from omegaconf import DictConfig, OmegaConf, SCMode

from db_connect_v2_image_classification.configs import AppConfig


def main_wrapper():
    # prepares the config store for any entrypoints that will use this decorator
    # since we're using the same AppConfig both for table creation.
    # for frontend app, we'll simply re-use this function.
    cs = ConfigStore.instance()
    cs.store(name="base_config", node=AppConfig)

    # a decorator that will wrap any function under @main_wrapper()
    def decorator(func: Callable[[AppConfig], None]):
        # hydra.main to enable flexible config provisioning
        @hydra.main(version_base=None, config_name="config")
        def _wrapped(cfg: DictConfig):
            # important - we're using dataclasses with additional functionality inside them
            # (e.g. methods or __post_init__).
            # by default hydra returns an untyped DictConfig.
            # with this we convert the config back into a Python dataclass with all methods.
            _cfg = OmegaConf.to_container(cfg, structured_config_mode=SCMode.INSTANTIATE)
            func(_cfg)

        return _wrapped

    return decorator

Whoosh, here things get a bit of complex (classics with Python decorators). Let’s schematically describe this logic. What we want is a flexible wrapper for an entrypoint function:

This wrapper should be easy to apply to various entrypoint functions (we’ll have two of them — one for table creation, another one for frontend app launch).

What is the easiest way to add logic to functions in Python? Exactly — decorators. Our code would look like this:

A profound reader will also notice that we’ve added a cfg argument to the entrypoints. This is done so we could easily work with the configuration without reading it separately in separate parts of the application.

One more thing is the way how hydra (and omegaconf which is running under the hood of it) handles the structured configs:

@hydra.main(version_base=None, config_name="config")
        def _wrapped(cfg: DictConfig):
           # work with cfg object

It actually doesn’t instantiate the dataclass itself, simply providing a dict config after structure validation. To get back the real object with proper methods, a small trick is required:

@hydra.main(version_base=None, config_name="config")
def _wrapped(cfg: DictConfig):
    # important - we're using dataclasses with additional functionality inside them
    # (e.g. methods or __post_init__).
    # by default hydra returns an untyped DictConfig.
    # with this we convert the config back into a Python dataclass with all methods.
    _cfg = OmegaConf.to_container(cfg, structured_config_mode=SCMode.INSTANTIATE)
    func(_cfg)

With all this ✨arcane knowledge✨ we can now wrap this into a nice decorator (code provided above), and use it without worries through the whole application.

⬆️ Populating the image metadata table

It’s time to populate the image metadata table. The code is provided below:

# db_connect_v2_image_classification/create_table.py

from databricks.connect.session import DatabricksSession as SparkSession
from databricks.sdk.core import Config
from pyspark.sql.functions import element_at, split

from db_connect_v2_image_classification.configs import AppConfig
from db_connect_v2_image_classification.main_wrapper import main_wrapper


class CreateTableTask:
    def __init__(self, cfg: AppConfig):
        self.cfg = cfg
        self.spark = SparkSession.builder.sdkConfig(
            Config(profile=self.cfg.profile, cluster_id=self.cfg.cluster_id)
        ).getOrCreate()

    def launch(self):
        print(f"Loading image metadata into the table {self.cfg.image_table}")
        print(f"Using cluster {self.cfg.cluster_name} with id {self.cfg.cluster_id}")

        metadata_df = self._load_image_metadata()
        metadata_df.write.saveAsTable(f"{self.cfg.image_table}", format="delta", mode=self.cfg.table_saving_mode)
        print(f"Image metadata has been successfully saved into the table {self.cfg.image_table}")

    def _load_image_metadata(self):
        images = (
            self.spark.read.format("image")
            .load("/databricks-datasets/flower_photos/*/*.jpg")
            .select("image.origin")
            .withColumn("class", element_at(split("origin", "/"), 4))
            .withColumn(
                "image_id",
                element_at(split(element_at(split("origin", "/"), -1), ".jpg"), 1),
            )
            .select("image_id", "class", "origin")
        )
        return images


@main_wrapper()
def entrypoint(cfg: AppConfig):
    task = CreateTableTask(cfg)
    task.launch()

It’s a fairly simple piece of Spark code it terms of ETL — we load data from a default dbfs:/databricks-datasets folder (available in every Databricks workspace by default), and simply put the metadata into a separate table:

The interesting part is in the way how we initialize the Spark Session. Since we’re using DB Connect “V2”, it’s a fairly simple operation (snippet from the code above):

class CreateTableTask:
    def __init__(self, cfg: AppConfig):
        self.cfg = cfg
        self.spark = SparkSession.builder.sdkConfig(
            Config(profile=self.cfg.profile, cluster_id=self.cfg.cluster_id)
        ).getOrCreate()

This initializes a connection where DB Connect “V2” client is running on the local machine, and is capable of sending commands to the remote Databricks cluster.

🖼️ Adding a frontend app

For those who have had a chance to read my previous blogpost, this part would be pretty straightforward. This part of code initializes from an entrypoint function, passes the config and spins up the Dash app:

# file: db_connect_v2_image_classification/frontend/app.py

import logging
from random import choice

from dash import Dash, dcc, html

from db_connect_v2_image_classification.configs import AppConfig
from db_connect_v2_image_classification.frontend.css_utils import external_scripts, external_stylesheets
from db_connect_v2_image_classification.frontend.callbacks import prepare_callbacks
from db_connect_v2_image_classification.frontend.components import data_container, guideline, header, nav_container
from db_connect_v2_image_classification.frontend.crud import DataOperator
from db_connect_v2_image_classification.main_wrapper import main_wrapper

logger = logging.getLogger("frontend_app")
logger.setLevel(logging.INFO)


def prepare_layout(op: DataOperator) -> html.Div:
    logger.info("Preparing the layout")
    layout = html.Div(
        children=[
            header,
            guideline,
            data_container(op.get_all_classes()),
            nav_container,
            dcc.Store(id="current_index", data=choice(op.get_all_indexes())),
        ],
        className="p-5",
        style={"height": "100vh"},
    )
    logger.info("layout prepared")
    return layout


@main_wrapper()
def entrypoint(cfg: AppConfig):
    app = Dash(
        __name__,
        title="Image Classification App",
        external_scripts=external_scripts,
        external_stylesheets=external_stylesheets,
    )
    logger.info("Preparing the data operator")
    op = DataOperator(cfg)
    logger.info("Data operator loaded successfully")
    app.layout = prepare_layout(op)
    prepare_callbacks(app, op)
    app.run(debug=cfg.debug, host="0.0.0.0")


if __name__ == "__main__":
    entrypoint()

Line-by-line, the following happens:

1. Initialize an empty app object with some extra CSS (I’ve decided to add Tailwind and Daisy UI for some ready-to-use components).
2. Prepare the DataOperator instance. I’ve moved all backend-related actions (i.e. read/write ops) into this class for modularity.
3. Prepare the layout of the application
4. Prepare the callback actions
5. Serve the app

Let’s discuss some of the parts a bit deeper.

🔛 Backend interactions

As said above, the main backbone for R/W operations in this app is DB Connect “V2”. This allows users to write ETL-related code without wrapping it into an ORM-like solutions. For instance, here is the code relevant to the manipulations that could happen in the UI:

# file: db_connect_v2_image_classification/frontend/crud.py

import logging
from functools import lru_cache
from typing import List

import numpy as np
from databricks.connect.session import DatabricksSession as SparkSession
from databricks.sdk.core import Config
from pyspark.sql import DataFrame

from db_connect_v2_image_classification.configs import AppConfig


class DataOperator:
    def __init__(self, cfg: AppConfig):
        self.logger = logging.getLogger("frontend_app")
        self.cfg = cfg
        self.logger.info("Initializing DB Connect")
        self.spark = SparkSession.builder.sdkConfig(
            Config(profile=self.cfg.profile, cluster_id=self.cfg.cluster_id)
        ).getOrCreate()
        self.logger.info(
            f"DB Connect initialized a connection to cluster {self.cfg.cluster_name} with id {self.cfg.cluster_id}"
        )

    @property
    def _source_table(self) -> DataFrame:
        return self.spark.table(f"{self.cfg.image_table}")

    @lru_cache(maxsize=10_000)
    def get_all_indexes(self) -> List[str]:
        self.logger.info("Loading all image indexes")
        indexes = self._source_table.select("image_id").distinct().toPandas()["image_id"].to_list()
        self.logger.info("Image indexes loaded")
        return indexes

    @lru_cache(maxsize=10_000)
    def get_all_classes(self) -> List[str]:
        return self._source_table.select("class").distinct().toPandas()["class"].to_list()

    def get_image_class(self, image_id: str) -> str:
        return self._source_table.select("class").where(f"image_id = '{image_id}'").toPandas().loc[0, "class"]

    def get_image_payload(self, image_id: str) -> np.ndarray:
        image_origin = self._source_table.select("origin").where(f"image_id = '{image_id}'").toPandas().loc[0, "origin"]
        image_info = self.spark.read.format("image").load(image_origin).toPandas().T.squeeze()

        img_payload = np.frombuffer(image_info["data"], dtype=np.uint8).reshape(
            image_info["height"], image_info["width"], image_info["nChannels"]
        )[:, :, ::-1]

        return img_payload

    def update_image_class(self, image_id: str, new_class: str):
        self.logger.info(f"Updating the image class for image {image_id}")
        command = f"UPDATE {self.cfg.image_table} SET class='{new_class}' WHERE image_id='{image_id}'"
        df = self.spark.sql(command)
        df.collect()
        self.logger.info("Update finished")

In the initialization phase we connect to a Databricks cluster via DB Connect, then we can add read-related methods in pure Spark APIs.

However, some fiddling is required to properly put the binary image data into plotly.imshow :

def get_image_payload(self, image_id: str) -> np.ndarray:
        image_origin = self._source_table.select("origin").where(f"image_id = '{image_id}'").toPandas().loc[0, "origin"]
        image_info = self.spark.read.format("image").load(image_origin).toPandas().T.squeeze()

        img_payload = np.frombuffer(image_info["data"], dtype=np.uint8).reshape(
            image_info["height"], image_info["width"], image_info["nChannels"]
        )[:, :, ::-1]

        return img_payload

Line-by-line on what’s happening:

1. By running a select on the metadata table, function retrieves the image path (origin) in the cloud storage
2. The image information is loaded into a single-element of pd.Series
3. The binary data together with height/width and channels information can be used to build a numpy ndarray
4. As a last step, this array needs to be reversed to fit the RGB order (this is where we apply [:,:,::-1] operation

A nice cherry on the cake is that it’s still possible to use old good SQL, e.g. for the UPDATE operations:

def update_image_class(self, image_id: str, new_class: str):
        self.logger.info(f"Updating the image class for image {image_id}")
        command = f"UPDATE {self.cfg.image_table} SET class='{new_class}' WHERE image_id='{image_id}'"
        df = self.spark.sql(command)
        df.collect()
        self.logger.info("Update finished")

With this class, it’s now fully possible to cover any UI operation. Let’s bind these methods to the UI buttons:

# file: db_connect_v2_image_classification/frontend/callbacks.py

from random import choice

import plotly.express as px
from dash import Dash, Input, Output, State

from db_connect_v2_image_classification.frontend.crud import DataOperator


def prepare_callbacks(app: Dash, op: DataOperator):
    @app.callback(
        Output("current_index", "data"),
        Output("image_id", "children"),
        Output("class_selector", "value"),
        Output("img_display", "figure"),
        Input("random_btn", "n_clicks"),
    )
    def get_next_random(_):
        next_index = choice(op.get_all_indexes())
        img_class = op.get_image_class(next_index)
        img_payload = op.get_image_payload(next_index)
        figure = px.imshow(img_payload)
        figure.update_layout(coloraxis_showscale=False)
        figure.update_xaxes(showticklabels=False)
        figure.update_yaxes(showticklabels=False)
        return next_index, f"Image id: {next_index}", img_class, figure

    @app.callback(
        Output("output_mock", "children"),
        Input("confirm_btn", "n_clicks"),
        State("class_selector", "value"),
        State("current_index", "data"),
    )
    def save_selected_class(_, value, current_index):
        if value:
            op.update_image_class(current_index, value)
        return value

It turns out that the app has a pretty simply logical nature — we only have a random image picker (which will update necessary outputs on each “Next” click), as well as a class saver (which will be triggered on a “Confirm” click.

🔌 Adding UI components

Since the app itself is pretty simple, I’ve only used a couple of nice components and capabilities of Tailwind CSS and Daisy UI:

# file: db_connect_v2_image_classification/frontend/components.py

from dash import dcc, html

blogpost_link = (
    "https://www.databricks.com/blog/2022/07/07/introducing-spark-connect-the-power-of-apache-spark-everywhere.html"
)

header = dcc.Markdown(
    "## Image classification app, built with [Dash](https://plotly.com/dash/) "
    f"and [DB Connect V2]({blogpost_link}) 🔥",
    className="text-3xl mb-2",
)


guideline = dcc.Markdown(
    "Please click below to navigate between images and correct their class labels when required.",
)

random_btn = html.Button(
    "🔀 Next image",
    id="random_btn",
    n_clicks=0,
    className="btn btn-primary btn-lg mt-5",
)

nav_container = html.Div(children=[random_btn], className="flex justify-center")

confirm_button = dcc.Loading(
    id="submit-loading",
    children=[
        html.Button(
            "Confirm",
            id="confirm_btn",
            n_clicks=0,
            className="btn btn-success btn-block my-4",
        ),
        html.Div(id="output_mock", style={"display": "none"}),
    ],
)


def class_selector(class_choices):
    return html.Div(
        children=[
            html.Div(
                children=[
                    dcc.Loading(
                        children=[
                            html.P("Please choose the class below:"),
                            dcc.Dropdown(
                                class_choices,
                                id="class_selector",
                                className="dropdown-content text-black",
                                placeholder="Select the class",
                                multi=False,
                                clearable=False,
                            ),
                        ]
                    )
                ],
            ),
            confirm_button,
        ],
    )


def data_container(class_choices):
    return html.Div(
        children=[
            html.Div(
                className="flex justify-center pt-2",
                children=[
                    html.Div(
                        className="card w-96 bg-base-100 shadow-xl p-4",
                        children=[
                            dcc.Loading(
                                [
                                    dcc.Graph(id="img_display"),
                                    dcc.Markdown(id="image_id", className="text-sm text-sky-500"),
                                ]
                            ),
                            html.Div(className="divider"),
                            class_selector(class_choices),
                        ],
                    )
                ],
            ),
        ]
    )

The one I’ve liked the most is the ready-to-use card class with all relevant presets. Take a look at how with just a couple of lines of CSS classes the app could become vivid and UI-friendly:

⚙️ Combining it all together

Finally, it’s time for a quick demo of the application:

With this app users can easily classify the images, without a need to use an external tool. At the same time, all data is stored in the cloud storage, whilst metadata is saved and updated in a Delta Table.

✅ Summary

With the flexibility of the Dash framework, it’s possible to build modern-looking and flexible frontend applications.

At the same time, DB Connect works in the backbone of the application, providing convenient Spark APIs to the client-side for proper operations with any kind of data (e.g. images).

The image data itself can be stored in the cloud storage, whilst the metadata information can be easily retrieved from and updated in the Delta Table.

All these technologies together provide a powerful toolset for building efficient and modern data applications on top of the Databricks Lakehouse, in pure Python and with the convenience of the well-known APIs.

The source code from this blog post is provided here. Feel free to copy it and fiddle around with your use cases 🙌.

Have you tried using DB Connect “V2” and Dash already? Have an opinion? Feel free to share it in the comments! Also, hit subscribe if you liked the post — it keeps the author motivated to write more.

In today’s world, we generate and process a vast amount of data every day, and it’s becoming increasingly challenging to manage it all efficiently. Future-proof data architectures that can handle large volumes of data are essential for businesses and organizations.

In this blog post, I would like to highlight how Apache Spark and its solid set of APIs can ease some technical burdens and make the life of Data Engineers slightly better.

N.B. This post is experimental and doesn’t contain the full code version, merely the snippets and concepts.

The metaphysics of data

First, we shall consider the concept of data itself and the directions of usage. If you’ll check any relevant source, you’ll find something like this:

Unfortunately, it’s not provable on a big scale, but I tend to think that the raw data itself is most usually not the main driver of growth in size.

It grows when it gets combined and reshaped.

In my practice I see that the most valuable data is forged out of the following:

• Several sources are combined (e.g. enriched or cross-checked)
• Data appears in time for the business to make a decision
• Data is properly transformed and quality assured

On the implementation side, we have pipelines that have a goal to deliver data under the requirements described above.

And the Data Engineers who are writing these pipelines usually want efficiency in their day-to-day work, for instance:

• Write expressive and efficient code
• Write code that is extensible enough to catch future use cases with ease
• Write code that will be reliable and won’t wake up you in the middle of the night with an alert

Engineering knowledge tends to be forged and framed out of the fact that people are solving the same problems in different domains — and this is how frameworks are born.

In the data space, it’s common to use the OSS data processing frameworks (aka ETL frameworks), and one of the most adopted data frameworks in the world is called Apache Spark.

And I believe amongst other reasons for its incredible adoption, there are 3 of the most noticeable ones:

• Abstractions when needed
• Scalability
• Flexible scheduling

Combination of which adds the future-proof label to the architecture of data applications. Therefore I would like to highlight these capabilities to the audience.

Abstractions when needed

Apache Spark allows you to write code without being bound to the specifics of the technical details to a greater extent.

By this, I mean that it becomes possible to write heavily domain-oriented code with Spark APIs being used when needed, without enforcing users to build everything around the framework.

Let’s consider the following example: I have a process that reads raw IoT data from a source. I want to apply some changes to the input, join it with some other data and save the result somewhere.

It is technically possible to make it like this:

if __name__ == "__main__":
    raw_iot_events = spark.read.format("some").load(source_path)
    devices = spark.createDataFrame(pd.read_csv(devices_path))

    iot_with_devices = raw_iot_events.join(devices, on="device_id")
    cleaned_iot = iot_with_devices.where("value > 1")
    cleaned_iot.write.somewhereElse()

And this is a valid code. However, it’s not that nice from the perspective of the future, due to several reasons:

• It doesn’t contain any proper domain model
• It doesn’t abstract the physical details
• And there are at least two more issues considered later in this post.

Let’s rewrite it to something that will be actually more domain-driven:

class Pipe:

  def source(self, source_definition):
    ...
  
  def destination(self, destination_definition):
    ...

  def join(self, join_rule):
    ...
  
  def transform(self, transformer):
    ...


class LaunchScheduler:
  ...
  

class IotProcessingTask:

    def launch(): # application layer - no physical details here
        pipe = (
            Pipe()
                .source(RawEvents(conf.events)) 
                .source(RawDevices(conf.devices))
                .join(EventsAndDevices)
                .transform(EventsCleaner)
                .destination(CleanEvents)
        )
        pipe.launch_with(LaunchScheduler(conf.launch))

if __name__ == "__main__":
    task = IotProcessingTask()
    task.launch()

I’ve intentionally omitted the implementation details to underline the flexibility of the solution and its independence from the physical implementation (because the framework is the physical detail and not the domain-related one).

Why do this? Because we want our code to be properly abstracted for the future (in this context it means stable to the external changes).

Let’s consider that format of the input is an axis of potential changes. In this case, we’ll need simply to change the properties of the configuration, no code changes are required.

For instance, we can implement a class that is populated from the conf in runtime (it can be even a pydantic BaseModel):

class TableCoordinates(BaseModel):
  catalog: Optional[str]
  database: Optional[str]
  table: str

class RawEvents(BaseModel):
  source_type: Literal["file", "table"]
  source_path: Optional[str]
  source_format: Optional[str]
  source_table: Optional[TableCoordinates]

Into these classes, we can easily pass any kind of payload from a static YAML/JSON/another config format, without a need to change a single line of code:

Another potential axis of change is the logic of how we work within the entity. For example, we got a new field that we need to handle in some specific way:

class RawEventsTransformer: # add it as a mixin to the application layer
    def transform(df):
        return df.withColumn("event_id", F.cast("event_id", "bigint"))

This change is fully isolated from the rest of the codebase, leading us to loosely coupled components that are simple to reuse further.

Scalability

Now we’re going into the next dimension of the data pipelines, namely the scalability topic.

There have been a lot of discussions about using libraries for single-node tasks in the ETL chain. While I appreciate the effort to explore new solutions, I often find myself questioning this approach.

My reason to worry is that it’s a “working now” solution instead of a future-proof solution.

# this goes into OOM as soon as source significatly grows in size
some_source = single_node_package.read_csv(some_source_path)
transformed_in_single_node = some_single_node_transforms(some_source)

...
df_in_spark = spark.createDataFrame(transformed_in_single_node)

As I’ve pointed out in the metaphysics part — data has a tendency to grow.

How soonish could it happen that the some_source the dataset will become out of memory? What would happen if we’ll simply add more rows, or change the logic of the dataset population, e.g. add explicit versioning of the records in this source?

The solution above will blow out with an OOM, leading developers to incidents and a need to refactor this piece. Indeed there are workarounds, e.g. upscaling the single node to the acceptable RAM amount though. But why would we introduce tech debt that requires workarounds?

Instead, there is a solution that will be scalable by default:

some_source = spark.read.format("some_format", some_source_path)
transformed_in_distributed = distrubuted_transforms(some_source)

And even more, with the Pandas API on Spark transformations in Pandas are available inside Spark itself. Moreover, any code that would accept and produce Pandas DataFrames can be used within Spark itself with Pandas UDFs.

Flexible scheduling

One important feature of Spark is the flexible API to launch the ETL code in various processing frequency circumstances.

E.g. you might want to write your code once and make it useful both in batch and streaming, with the capability to easily switch between these behaviors.

Remember LaunchScheduler from the beginning? Let’s take a closer look at the potential design of this part:

LaunchScheduler(BaseModel):
  once: Optional[bool]
  available_now: Optional[bool]
  processing_time: Optional[str]


Pipe:
  def launch_with(scheduler_config):
    # do the basics first
    sources = [spark.readStream(source) for source in self.source_definitions]
    sources_transformed = [self._transform_source(source) for source in sources]
    joined = self._join_with_rule(self.join_rule)
    
    # process inside batch / micro-batch
    def batch_processor(df, epoch_id):
       df = self._post_join_transforms(df)
       self._write_to_destination(df) # encapsulates writing logic

    query = (
        spark
            .foreachBatch(batch_processor)
            .trigger(scheduler_config)
            .start()
    )
    query.awaitTermination()

The brilliant part comes in the following — it becomes possible to write the ETL code once.

When a batch is required, it’s enough to useonce or available_now (read here on the differences), and when it comes to a business requirement for faster SLAs, it’s easy to switch to other triggers without changing the code itself, and by just changing the configuration.

Summary

Apache Spark as a framework allows users to build apps that can be easily and efficiently scaled. At the same time code of the data application can be domain-driven and flexibly scheduled without a need to develop two versions of code for batch and streaming use cases.

These capabilities combined together allow developers to deliver their data applications in a future-proof way without the requirement of costly refactoring in case of changed requirements and data growth.

Have you tried using writing data applications with Apache Spark in a domain-driven design? Have an opinion? Feel free to share it in the comments! Also, hit subscribe if you liked the post — it keeps the author motivated to write more.

✍️ Intro

Data labeling is a crucial aspect of machine learning that involves annotating raw data to create labeled datasets that can be used to train models.

However, the process of labeling large volumes of data can be time-consuming and error-prone. Fortunately, there are several tools and technologies available that can simplify the data labeling process and increase its efficiency. In this blog post, we will explore how to build a data labeling app using Python, Dash, and the Databricks Lakehouse platform.

By the end of this post, you will have a better understanding of how to leverage these powerful tools to create a custom data labeling app that can save you time and increase the accuracy of your labeled datasets.

🏗️ Architecture

Let's take a closer look at the architecture of the labeling app. As usual, we have several layers — frontend, backend, and storage:

Some considerations that make this approach distinct from others:

• [Frontend] Data Scientists usually would prefer to stick to the main language of their choice, which is usually Python. Therefore we’ll use Dash — a robust framework for building UI applications in pure Python.
• [Backend] This time we’ll use Databricks SQL not as a high-performant query backend (as it’s used in many BI apps), but as a CRUD interface to the table with data.
• [Storage] Since DBSQL is being used and we’ll require Updates, Delta is an obvious choice.

To sum up — we’re going to build a CRUD-like application with UI on top of the data stored in the Databricks Lakehouse.

❓What will this app do?

The app described in the blog post implements a user interface to classify text blocks. Users will be able to open the app in the browser, see the text blocks and select the relevant class per each block. By clicking confirm, the newly selected class will be saved back into the Lakehouse.

🔷 Frontend

First, let’s quickly sketch out the UI component. What we’re going to build is a simple app that allows the user to see some text and then classify it to a given label:

Well, maybe it’s not the most enjoyable UI, but definitely a simple and clear one. Hopefully, professional web designers can forgive me for the UX of this one.

If you have ideas on how such an app should look like, drop a line in the comments!

Now it’s time to code a bit. I’ll omit the package setup, take a look at the developer instructions in this repo for details.

Let’s start by sketching out the layout:

app.layout = html.Div(
    [
        html.Div(
            [
                header,
                guideline,
                current_index_view,
                html.Div(
                    [
                        text_container,
                        html.Div(
                            [
                                class_guideline,
                                dropdown(CLASSES),
                                confirm_button,
                            ],
                            style={"flex-basis": "10%", "padding-left": "1em"},
                        ),
                    ],
                    style={
                        "display": "flex",
                        "flex-direction": "row",
                        "padding": "2em",
                        "margin-bottom": "2em",
                    },
                ),
                html.Div(
                    navigation_buttons,
                    style={
                        "justify-content": "space-evenly",
                        "display": "flex",
                    },
                ),
                dcc.Store(id="current-index", data=choice(ALL_IDS)),
            ],
            style={
                "padding-top": "1em",
                "padding-left": "2em",
                "padding-right": "2em",
                "padding-bottom": "5em",
            },
        ),
    ]
)

I did some minor changes to the outlook of the original sketch. Let’s see what comes out of it:

Well, this looks kinda functional, but we definitely miss some 💫fancy💫 feeling here.

Let’s fix these things and sugar-coat it a bit with CSS. We’re going to do several things:

• Add bootstrap to make styles nicer
• Switch to another font
• Add a nice background gradient (because it’s cool and easy!)

Adding Bootstrap support is pretty easy:

external_stylesheets = [
    # bootstrap
    {
        "href": "https://cdn.jsdelivr.net/npm/bootstrap@3.4.1/dist/css/bootstrap.min.css",
        "rel": "stylesheet",
        "integrity": "sha384-HSMxcRTRxnN+Bdg0JdbxYKrThecOKuH5zCYotlSAcp1+c8xmyTe9GYg1l9a69psu",
        "crossorigin": "anonymous",
    },
    # fonts
    {"rel": "preconnect", "href": "https://fonts.googleapis.com"},
    {
        "rel": "preconnect",
        "href": "https://fonts.gstatic.com",
        "crossorigin": "anonymous",
    },
    {
        "href": "https://fonts.googleapis.com/css2?family=Inter&family=Noto+Sans&display=swap",
        "rel": "stylesheet",
    },
]

# later in the app - add a reference
app = Dash(
    __name__, title="Data Labeling App", external_stylesheets=external_stylesheets
)

Note: I’m using Bootstrap 3 which is pretty old. Consider using the latest version.

To make buttons look nice, simply add relevant class names to them:

# later in the layout - add some classes to make buttons look better
confirm_button = dcc.Loading(
    id="submit-loading",
    children=[
        html.Button(
            "Confirm",
            id="confirm_btn",
            n_clicks=0,
            className="btn btn-success btn-block btn-md",
        ),
        html.Div(id="output-mock", style={"display": "none"}),
    ],
)

All these steps will make our app look like this:

This already looks softer, but I really don’t like the default fonts of Bootstrap.

Therefore, let’s also add a new folder called assets and put a file custom.css into it. Please note the package layout:

|-- README.md
|-- dbsql_labeling_app_example
|   |-- __init__.py
|   |-- app.py
|   |-- assets
|   |   `-- custom.css

The assets folder should be in exactly the same folder as the app.

With assets configured, put the font settings into the custom.css file:

body {
    font-family: 'Inter', sans-serif;
    height: 100%;
}

html {
    height: 100%;
}

h1 {
    margin: 0
}

a {
    color: plum;
}

However, the result looks a bit strange — the whole page is out with just one single font. What would be nice is to logically separate the text block that users will classify from the rest of the app.

This can be achieved by providing a custom font for a component:

# note the styling part with font-family.
dcc.Markdown(
                id="text-container",
                style={
                    "font-size": "1.2em",
                    "font-family": "'Noto Sans', sans-serif",
                    "height": "50vh",
                    "overflow-y": "auto",
                },
            ),

All this together will end up in the following UI:

This already looks slicker, but still not 💫fancy💫. To ensure the proper level of fanciness a nice background gradient is definitely required. I’ve used the ColorSpace tool to generate a cool gradient and put it into the body definition:

body {
    background-image: linear-gradient(to right top, #1b3139, #164472, #754293, #d41677, #ff3621);
    color: whitesmoke;
    font-family: 'Inter', sans-serif;
    height: 100%;
}

All this together will lead to the following:

The styling is unfortunately only one part of the frontend story. The second one is logic that connects buttons and web components. The logic here is a bit non-linear, therefore writing down the requirements would be helpful.

The app should be able to show any of the texts provided by backend, together with their associated labels. User should be able to change the label of the text, and should be provided with choices of available classes.

This description leads to the following:

• A state with an index is required. The index will be changed via the buttons and based on the index the text and label will be demonstrated.
• A variant selector is required. When the user confirms the choice, the chosen variant should be sent to the backend via update-by-id logic.

Fortunately, Dash has great capabilities for state management and callbacks.

The index part is trivial with the provided capabilities. First, it’s required to define buttons — they’ll will act as inputs:

navigation_buttons = [
    html.Button(
        "⬅️ Previous",
        id="prev_btn",
        n_clicks=0,
        className="btn btn-default btn-lg",
    ),
    html.Button(
        "🔀 Random",
        id="random_btn",
        n_clicks=0,
        className="btn btn-default btn-lg",
    ),
    html.Button(
        "Next ➡️",
        id="next_btn",
        n_clicks=0,
        className="btn btn-default btn-lg",
    ),
]

Second, let’s introduce the index as a store component:

dcc.Store(id="current-index", data=choice(ALL_IDS)),

The Store component allows you to store the data that can be JSON-serialized inside the browser (somewhat similar to React useState).

Finally, it’s time to bind the buttons to the index. In Dash such interactions are organized as callbacks:

@app.callback(
    Output("current-index", "data"),
    Output("text-container", "children"),
    Output("class-selector", "value"),
    Output("prev_btn", "disabled"),
    Output("next_btn", "disabled"),
    inputs=[
        Input("current-index", "data"),
        Input("prev_btn", "n_clicks"),
        Input("next_btn", "n_clicks"),
        Input("random_btn", "n_clicks"),
    ],
)
def navigate_to_element(current_index, _, __, ___):
    if "prev_btn" == ctx.triggered_id:
        current_index -= 1
    elif "random_btn" == ctx.triggered_id:
        current_index = choice(ALL_IDS)
    elif "next_btn" == ctx.triggered_id:
        current_index += 1

    label_data = operator.get_element_by_id(current_index)
    disable_previous = current_index <= 0
    disable_next = current_index >= len(ALL_IDS) - 1
    return (
        current_index,
        label_data.text,
        label_data.label,
        disable_previous,
        disable_next,
    )

This specific callback gets triggered in case if any of the navigation buttons are clicked:

The outputs define the side effects of these clicks. As per the code above, there are 5 potential side effects:

• The current index itself will be changed
• Text container component will be filled with text of the relevant sample that is fetched from the Lakehouse
• The label component will be filled with the label of the relevant sample
• Two last outputs will disable Previous/Next buttons if there is no available Id.

Using the same logic, it’s simple to organize the class choice and confirmation:

From the UI perspective, two components are introduced:

def dropdown(classes: List[str]) -> html.Div:
    return html.Div(
        dcc.Loading(
            id="dropdown-loader",
            children=[
                dcc.Dropdown(
                    classes,
                    id="class-selector",
                    placeholder="Select the class",
                    multi=False,
                    clearable=False,
                    style={
                        "color": "black",
                    },
                ),
            ],
        ),
        style={
            "padding-top": "10px",
            "padding-bottom": "10px",
        },
    )


confirm_button = dcc.Loading(
    id="submit-loading",
    children=[
        html.Button(
            "Confirm",
            id="confirm_btn",
            n_clicks=0,
            className="btn btn-success btn-block btn-md",
        ),
        html.Div(id="output-mock", style={"display": "none"}),
    ],
)

And then a callback is added to bind events together:

@app.callback(
    Output("output-mock", "children"),
    Input("confirm_btn", "n_clicks"),
    State("class-selector", "value"),
    State("current-index", "data"),
)
def save_selected_class(_, value, current_index):
    if value:
        operator.update_element_by_id(current_index, value)
    return value

A profound reader might notice above an interesting twist — what’s that output Output(“output-mock”, “children”) is actually used for?

The explanation is fairly simple — Dash doesn’t support callbacks without outputs.

At the same time, it’s possible to create a dummy output object that is not visible to users:

html.Div(id="output-mock", style={"display": "none"}),

Moreover, this object will become pretty useful further when a Loading state will be introduced.

The loading state handling is a topic that is covered with Dash built-in capabilities completely:

text_container = html.Div(
    dcc.Loading(
        id="text-block-loading",
        children=[
            dcc.Markdown(
                id="text-container",
                style={
                    "font-size": "1.2em",
                    "font-family": "'Noto Sans', sans-serif",
                    "height": "50vh",
                    "overflow-y": "auto",
                },
            ),
        ],
    ),
    style={
        "flex-basis": "90%",
        "padding-right": "1em",
    },
)

Hint: put a html.Div with generic layout styles on top of Loading component so it will look stable in the UI.

And this is it — no need to connect callbacks, if-else, when, etc. Simple and concise wrapper around the UI component that’s being loaded.

Now to the cool tricks — the output-mock we’ve used above can be used to add a Loading state to the confirm button itself:

confirm_button = dcc.Loading(
    id="submit-loading",
    children=[
        html.Button(
            "Confirm",
            id="confirm_btn",
            n_clicks=0,
            className="btn btn-success btn-block btn-md",
        ),
        html.Div(id="output-mock", style={"display": "none"}),
    ],
)

Since the button is not reloaded when it’s clicked, it’s possible to simply wrap it into a Loading block together with its mocked output. This will add a smooth confirm click behavior:

The app is full of all required utilities for text labeling, and the only thing it awaits is the backend connection to it.

💎 Backend

Per the code above the logic of the backend seems pretty obvious. To simplify the CRUD-related operations I’ve put all of the relevant methods into a single class:

from itertools import chain
from typing import List, Optional

from sqlalchemy import select
from sqlalchemy.orm import Session

from dbsql_labeling_app_example.engine import Label, engine


class DataOperator:
    def __init__(self) -> None:
        print("Initializing the session for CRUD operator")
        self._session = Session(bind=engine)
        print("Session initialized")

    def get_all_ids(self) -> List[int]:
        return self._session.execute(select(Label.label_id)).scalars().all()

    def get_element_by_id(self, id: int) -> Optional[Label]:
        return self._session.query(Label).get(id)

    def update_element_by_id(self, id: int, new_label: str):
        self._session.query(Label).filter_by(label_id=id).update(
            {Label.label: new_label}
        )

    def get_all_classes(self) -> List[str]:
        return list(chain(*self._session.query(Label.label).distinct().all()))

This class is used to provide data from the backend towards the frontend app (and therefore used in various callbacks).

The real beauty of SQLAlchemy is that all these standard CRUD operations are already in place.

At the same time, the Databricks SQL Connector can be easily used together with SQLAlchemy, providing efficient functions and methods to build CRUD operations of any complexity:

from sqlalchemy import create_engine

engine = create_engine(
        f"databricks://token:{endpoint_info.token}@{endpoint_info.server_hostname}?http_path={endpoint_info.http_path}&catalog={endpoint_info.catalog}&schema={endpoint_info.database}",
        echo=debug_mode,
    )

An additional bonus of using SQLAlchemy is that the code becomes well-typed. For instance, the objects that are used in the frontend code, as well as their attributes can be defined in a few lines of code:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

from dbsql_labeling_app_example.engine_provider import get_prepared_engine

print("Preparing the SQL engine")
engine = get_prepared_engine()
Base = declarative_base(bind=engine)
print("SQL Engine prepared")


class Label(Base):
    __tablename__ = "labels"

    label_id = Column(Integer, primary_key=True)
    text = Column(String)
    label = Column(String)

Since the frontend code shares the same codebase and runtime, SerDe between “backend” and “frontend” becomes obsolete.

✅ Summary

• CRUD with DBSQL and Delta directly on top of the Lakehouse is a real thing
• With the Dash framework, one can quickly build efficient data applications.
• SQLAlchemy and DBSQL work together pretty well, which allows reusing existing ORM tooling and brings a lot of benefits to Python users.

The source code for this app as well as demo can be found here. Feel free to copy it and fiddle around with your use cases 🙌.

Have you tried using Databricks SQL and Dash already? Have an opinion? Feel free to share it in the comments! Also, hit subscribe if you liked the post — it keeps the author motivated to write more.

Scala is an incredibly flexible programming language that enabled Software Engineers to build powerful and easily extensible applications in various domains.

Since a large chunk of Apache Spark is written in Scala, it’s also pretty natural to write Spark applications in Scala language to utilize its typing system fully.

Although Scala is well-known for its functional libraries (and code written in a pure-functional, monadic style is something that might look unusual for some developers), it also allows users to write in object-oriented style with minor additions of functional programming where needed.

In this blog post, I wanted to demonstrate some common usage patterns for writing Spark Scala applications in a concise, object-oriented style.

Using HOCON-based configurations

HOCON — is a very flexible configuration language, actively used in many Scala-based projects. It supports a pretty wide range of additions, most notable of which I find the following:

• Environment variable substitution
• In-place variable substitution (including nested configs)
• Comments
• Duration and size-like properties, e.g. “10 seconds”

Here is a simple example of writing and reading such a config:

Reading the config is trivial if we want a standard untyped Config object:

A nice feature of the HOCON is to share configs across various applications. For example, if you need to configure a chain of tasks, where inputs of one task are dependent on the output configuration of a previous one — HOCON can easily resolve such cases, e.g.:

Another important point of using HOCON is the ability to work nicely with tests. Later in this post, we’ll see how HOCON-based definitions could be injected with testing properties.

Simplifying tests with testcontainers-scala

As you can see in the configuration, we’re going to write an application that reads data from a Kafka topic. Therefore, we would need to either mock or provide an example of a Kafka stream in our local environment for proper local tests. Here we can easily use a great JVM-based tool called testcontainers, specifically an extension that makes this library easy to use in Scala.

By using scalatest and testcontainers developers can easily set up a proper environment for local testing:

Here is another useful example — this trait provides support for SparkSession with Delta Lake enabled for test cases:

Most certainly an attentive reader has already noticed that spark object is introduced as an implicit here. Therefore let’s talk a little bit about the application layout which makes the setup easier and more concise from the programming perspective.

The implicit for spark is used for the following reasons:

• It simplifies access to the SparkSessioninstance across the codebase
• Easy injection of this instance across the testing codebase

Following this idea, one can easily guess that it’s simple to introduce a package-level object which provides access to the SparkSession:

Now any application inside net.renarde.demos.apps will have access to the Spark Session instance, and in the testing environments a proper local session will be initialized.

Generic app trait and reading the configs

Since every application requires a config and a logger instance, a generic app trait can be introduced:

N.B. I’ve omitted ConfigProvider code for simplicity, take a look at the utils package for more details.

This trait is simple to re-use in various applications, e.g.:

By this, we avoid re-writing boilerplate code which reads configs and initializes the logging and SparkSession instance — providing a smooth and fast start to develop a new application.

Let’s now take a look at the usage of configs. In my example, it’s required to provide KAFKA_BOOTSTRAP_SERVERS environment variable to properly resolve the configuration object. This can be easily done inside tests by overriding the getGlobalConfig function in the ConfigProvider :

Instead of direct reading from the application.conf (which would be done in case of production launch), the configuration provider function is overridden with properties that are dynamically coming from the test environment configuration and then resolved with a fallback to the main configuration. This approach allows dynamical override and injection of the properties for proper unit testing.

Some examples and links

• An example project that uses these patterns with Kafka and Protobuf
• HOCON format docs
• testcontainers-scala library

If you know other useful patterns are there for Spark Scala application development, feel free to share them in the comments!

Also, hit subscribe if you liked the post — it keeps the author motivated to write more.

⏰ Borrowing time

Python is an extremely flexible language. And when I say extremely I sometimes mean it with a little grain of salt. It allows you to write absolutely untyped code without any worries.

For instance, when you would like to quickly parse an API response and simply extract the necessary results, you might be tempted to do it “quick and dirty”:

And in some cases, it’s a fine solution. Even more, it’s a valid Python code in the end. Tools like pylint won’t tell you anything bad about it.

However anyone who had a chance to write a bigger Python project will have immediate pain while reading this. They just feel that this bit of code will become an imminent source of bugs and issues in the future, especially when the project will start growing.

Stop for a second and consider — what kind of issues you might face with that code?

You got it right — literally every line in this code has a potential issue. Let’s describe some of them:

• No guarantees that request will be successful
• No guarantees that request is actually JSON compatible
• No guarantees that response payload will be a dict and not a list
• Even if it’s a dict, there are no guarantees that data key in it
• …

Same stuff applies to the pretty much every bit of the code above.

Someone would say — well, we have these guarantees from the API provider, right?

Yes and no. In fact these guarantees are well provided in case when “everything goes as expected” beyond your application or service boundary. But let’s be honest, this is a too wide statement to be true.

How could I as a developer regain at least minimal control over the input data, especially when it crosses the service boundary and unexpected structure becomes my problem? If I won’t enforce structures, I might run into:

def another_transform_func(inp: Dict[Any, Any]) -> Dict[Any, Any]:

Unfortunately my experience shows that using unverified structures on input, as well as Any->Any-like functions is a way to borrow time from future self. It allows you to move fast with the project, but will cost you bug-fixing time later on.

Is there any way to make it better?

🦢 The grace of pydantic

As a person who developed a lot in Scala, I indeed miss the case classes capabilities.

And to my opinion one of the best things that happened with Python ecosystem so far is indeed the development and huge adoption of pydantic. In fact, I don’t see any reason why it’s still not in the core Python instead of the built-in @dataclass functionality.

In the essence, pydantic allows you to create logical models of the entities in your Python application. This could be client/server models, this could be simple logical blocks that serve specific purpose — pretty much anything.

There are 3 absolutely great features that make pydantic stand out of the line, namely:

• Validators
• Nested Discriminated Unions
• Hypothesis plugin

👀 Validate ’em all

Validators are essential part of your code with pydantic models, since they allow you to verify the input data straight away at the moment of the object initialization. No need to see NoneType doesn't have get anymore — it will be covered just at the moment of object initialization.

Here are some of the nice tricks I use with pydantic:

This validation function allows you to verify that at least one field with specified suffix is provided — e.g. you expect that in the payload you have at least one field that should end with _name suffix. It’s easy to refer to this check in the class:

Another frequently used check is the mutual exclusivity of specific fields:

It could be easily applied in the following way:

In the end basic validation rules combined with structures could easily cover up to 100% of issues related to incorrectly parsed input data etc.

What’s more important is that it enforces usage of the logical models further down in your code. It’s way easier to understand a function when it’s clear that it receives one model and returns another one, rather than Any -> Any function.

🛂 Divide et impera

Another brilliant feature of pydantic is it’s capability to handle nested structures with various objects. It’s a pretty common case when you receive an input that contains various logical objects — and you would like to:

• Identify the object type
• And if the object type is not identifiable — throw an exception.

The example in the official doc demonstrates quite good the idea of this concept. However, it might be hard to add more custom logic to this identification process, e.g. identify object as A if previous object is B etc. Share your recipes below in the comments for that one!

🤔 What-if?

Another great capability is provided by the hypothesis plugin. Let’s be honest — no one wants to write sample models to test their code.

It’s just something that takes time and you’re kind of forced to specifically point your samples towards the errors you have (to improve test coverage indeed).

The hypothesis plugin makes your life slightly easier, and also adds more flexible scenario choices to your tests. Sometimes it helps to catch the “unexpected” errors, whilst manually written samples usually tend to verify the “expected” ones.

🐳 Summary

The necessity of using proper logical models is pretty clear in most of the cases. With the simplicity brought by pydantic, it makes sense to actually start writing new applications from logical models, and only then move to the application logic, organising it as a “dance of entities” (yep, it’s a reference to the Clean Architecture).

There are some small things that I would like to see better done in pydantic, for instance:

• generated fields straight away in the class definition
• custom functions for discriminated union instead of just property matching
• ability to mixin multiple root validators

But overall the experience using this library is that it brings efficiency and better code structure, as well as prevents errors.

Did you use pydantic in your projects before? Have an opinion? Feel free to share it in the comments! Also, hit subscribe if you liked the post — it keeps the author motivated to write more.

MLOps is a hot topic nowadays. Whilst the importance and business value of it is well described in various papers, e.g. [1], [2], there seems to be a lack of hands-on examples with explanations on how this could be done.

In this series of blogposts I’m going to demonstrate an example on how to build an ML application following MLOps practices and using Azure Databricks, Azure AKS and Azure EventHubs services. This part is dedicated towards model training and all the small details around it.

Process overview

To better understand the process, let’s take a look at this diagram:

This diagram tells us that we have three main components of our model that we shall take care of — code, model and data. This is something we’ll take care of with Azure Databricks platform. As for deployment and monitoring, we’ll also use Azure AKS and Azure EventHubs services.

High-level architecture

Let’s start from outlining the solution architecture in terms of used technologies and their roles:

As per MLOps standards, we need to keep jointed versioning of the following parts:

• Data — versioned in the Lakehouse, since it’s stored in Delta Lake Format with it’s Time Travel capability
• Code — versioned in GitHub
• Model — versioned in MLflow

MLflow also provides a model registry, which allows users to mark specific experiment runs as a model version and easily share the model to the serving layer (Azure AKS).

As for monitoring, we’re going to serve the model as a ReST API, which means we’ll need to collect the inputs and outputs of the model in real time. A good approach towards this is to send input-output pair from the serving interface directly towards Azure EventHubs, and then collect these metrics into a Lakehouse table. On top of this Lakehouse table we’ll be able to compute and visualize the metrics using DBSQL.

Preparing the project

Now it’ s getting to the technical part of implementation, time to roll up the sleeves and write some code. I’m going to use dbx since I prefer to develop ML pipelines in an IDE rather than in a notebook.

Since IDE is going to be used, some setup on the local machine is required:

1. Install conda
2. Create a new einvironment

conda create -n e2e-mlops-demo python=3.9

3. Activate a new environment:

conda activate e2e-mlops-demo

4. Install OpenJDK for local Spark tests:

conda install -c conda-forge openjdk=11.0.15

5. Install dbx:

pip install dbx

6. Create a new Databricks API token and configure the CLI:

databricks configure --profile e2e-mlops-demo --token

7. Check that profile is working correctly:

databricks --profile e2e-mlops-demo workspace ls /

Now the preparation is done, let’s generate a project skeleton using dbx init:

dbx init -p \
    "cicd_tool=None" \
    -p "cloud=Azure" \
    -p "project_name=e2e-mlops-demo" \
    -p "profile=e2e-mlops-demo" \
    --no-input

This command will generate the project template for us, with some useful settings and assets prepared.

8. Since I’m using GitHub, it’s easy to create repository from local machine and push it directly to the GitHub without using UI (this command is valid for MacOS and Linux, for PowerShell it needs to be slightly adopted):

gh repo create "${PWD##*/}" --private --source=. --remote=origin

9. Now it’s time to commit the initial version of the code:

git add . # add generated files to repo
git commit -m "init"
git push -u origin main

10. We also cal install and launch the project unit tests:

pip install -e ".[dev]"
pytest tests/unit --cov

I’ve curated the repository slightly (removed some irrelevant files), you can find the relevant version of it here.

Preparing the data

Before actually running any kind of a model, we’ll need to prepare some data. For demonstration purposes I’ve chosen the task of credit card fraud classification, and took the public dataset from OpenML.

Data is the one of the key parts of the whole MLOps process. On the Databricks Lakehouse platform we indeed store data in Delta Lake — efficient and fully open-sourced format which also supports versioning.

The code to download data into a Delta table is pretty much trivial (we simply download the DataFrame and do some conversions):

Main logic is defined in the launch method. We first prepare a separate database, then downloading the data into memory as a Pandas DataFrame, and do some transformations for convenience.

Final step is to convert this dataset to a Spark DataFrame and then save into a table. This table is also available for other users via SQL statements.

Data versioning

Since it’s a Delta Table, it also contains history of all operations:

We’ll use this feature in our ML training pipeline in the following fashion — when we read the table, we’ll check the latest version of it, and then log this metadata together with our model run into MLflow.

For better typing I’ve introduced logical models written in pydantic:

ModelData is the logical class which encapsulates train and test data, as well as providing the source metadata information (version, database and table).

The source metadata can be easily collected at a time of data reading:

Using Delta APIs it retrieves the last_version of the dataset, and then using Spark APIs it reads the dataset of a given version into a pandas DataFrame.

Noticing the imbalance

When it comes to the data, it’s easy to notice that this dataset could easily fit into memory:

It also seems like there is a huge class imbalance — pretty expected situation in fraud detection cases:

This minimal exploration provides us some valuable info:

• Non-standard metric for the classification model is required
• There is a potential for some over-sampling to balance classes
• There is no need for data or model parallelism, but we could speed-up the hyper-parameters search

Metric choice

Let’s start thinking about the metric first. There are many metrics that support the imbalanced learning, and one of them is Cohen’s Kappa. There are various discussions whenever it’s a best-in-class metric for imbalanced cases.

I’ve seen this metric being used a couple of times in real-world applications, and it’s already implemented in sklearn. You can read some relevant considerations and discussion here.

We’ll also calculate more standard F1 and ROC AUC scores for better introspection.

Smart over-sampling

To make training process more effective, I’m also going to add some smart oversampling from the great imbalanced-learn library.

Let’s take a look at the code that is responsible for re-balancing the dataset:

Although it looks a bit too verbose from the first sight, it’s not that scary in detail. After trivial split into X and y variables, we first split our data into train and test datasets with stratification by y variable.

Stratification is important to make sure we preserve the percentage of the target variable in both training and testing datasets.

Next step is to actually apply over-sampling technique, I’m using the ADASYN since it’s fast and pretty robust. Important to mention — we only generate the resampled data for training data, and we don’t touch the test to avoid test leakage.

At this point we’ve covered both topics of metric choice and over-sampling. However, we also have ability to scale out the process of hyper-parameter search. But there are so many scalability patterns, which one is to choose in this specific case?

Here I make a humble reverence to the scalability in ML tasks.

Different kinds of scalability

There are many various ways to scale the ML training process:

• Frameworks that support both data out-of-memory data and can scale with amount of computational resources —for example SparkML, H2O.ai, SynapseML.
• For deep learning, there are Horovod + Tensorflow/Pytorch.
• Most-popular boosting algorithms support running via Spark API, providing scalability both in terms of data size and compute(e.g. LightGBM, XGBoost, CatBoost)
• Another option is to parallelise training is to split the dataset horizontally and train a separate model per each partition. Partitioning key could be for example region, or segment. This functionality is available via Pandas UDF functionality and is well described here.
• If the dataset fits into memory, its possible to parallelise the process of hyper-parameter search. This functionality is covered by hyperopt integration.

All of these scalability patterns are available in the Databricks Lakehouse Platform.

Side note: check how many of these are available in the typical CDW scenario without exporting data out of it. After thinking about this, take a look at this blogpost from Jason Pohl.

Pipeline and search space

But let’s get back to the model building topic. For the purpose of this example I’m going to use the latter option and train my model using scikit-learn-based pipeline with hyperopt for hyperparameter search.

This approach requires to define the search space and the relevant model pipeline. I’ve factored out this code for better readability into a separate class:

As for model, I’m wrapping multiple steps into a chain using Pipeline API from sklearn. This approach unifies the model interface — now it becomes possible to replace the XGBClassifier with any other sklearn-compatible classifier and there is no need to fiddle around training interfaces.

Search space definition is pretty simple, it’s merely an example of how it could be configured. There is also one interesting thing about hyperopt API logic — after running the fmin function it will return a dictionary with a slightly different structure:

best_params = fmin(...)
print(best_params)

# example output:
{
    "max_depth": 3, 
    "n_estimators": 100,
    "learning_rate": 0.001
    "reg_alpha": 0.03
    "base_score": 0.02
}

It’s better to be aware of this when applying the best_params object to build the final model.

Training the model

With the considerations above it becomes pretty easy to factor out the code for model training into a separate class — Trainer . You can find the full code of this class in this repo, I’m going to point attention to some details of it.

To log parameters, metrics and the model in a convenient way a good idea is to create a structure of mlflow runs:

In the mlflow UI this structure will look like this, allowing ML engineers to manage and compare the runs in a simple fashion:

I would not recommend saving the model instance of each hyper-parameter run, although it’s possible and might be required in some cases.

From the API perspective, logging logic will look like this (follow the comments below):

• In the Trainer initialization step the experiment_id and the parent_run_id is obtained. No additional params or metrics are added in the run at this step, this will be done in the end of training process
• The entrypoint method train launches the hyperopt search with the self._objective method as function
• Inside the objective, self.setup_mlflow() method is called. This is only required for local testing, and this method will not have any effect when this code is launched on Databricks. For local testing it’s required to provide the MlflowInfo object during Trainer initialization— it will point the child process spawned by self._objective to use the local MLflow registry.
• The two-levelled with construction in _objective creates a nested run and provides context in which parameters and metrics will be saved by _train_model_and_log_results function.
• After running the fmin function, a dictionary of best_params is generated. At this point the model is re-trained once again with the best_params and then the model instance is saved into the parent run and registered into MLflow Model Registry as a new version.
• Point out the _register_model method which demonstrates how the package code is collected and stored together with the model object. Please note that the code_paths option is available in mlflow>=1.27.0 and on Databricks Runtime 11.0 ML and higher.

This concludes all the small details around model and code versioning. The main logic of the training is stored inside the _train_model_and_log_results method:

This function contains all the logic relevant to actually training and logging the model parameters and metrics. Whilst the pipeline.fit seems pretty standard for ML engineers, here are some additions in this code:

• standard sklearn method get_params() is pretty verbose when it generates parameters dictionary for Pipeline. Therefore I’ve slightly customized it, removing the steps section and auxiliary steps description from it.
• We explicitly save the source information to keep track of the source data version and location

Local testing

The provided Trainer class is easily testable on the local machine:

This test block runs the trainer with provided parameters and fixtures for dataset and mlflow_local instances. In the end it asserts that the model has been created and registered in the local temporary mlflow instance.

By writing a set of tests, it’s possible to cover the whole data loading and model training process with a decent amount of tests:

pytest tests/unit --cov --cov-report=html

There is indeed some place for improvement, but as a first shot it’s a good coverage metric.

Deploying the model to the Model Registry

The final step for this part is to train the model at scale and make it available in the Model Registry.

To do this, let’s define a Databricks workflow which consists of two steps:

This deployment file describes all the components we’ll need for this workflow, specifically:

• Cluster definitions
• Task definitions
• Relationship between tasks and clusters
• Order of execution between tasks

This file follows the Databricks Jobs API 2.1 specification, with some minor alterations that are features of dbx, for example:

• instance_pool_name and driver_instance_pool_name will be automatically replaced with their respective IDs
• Locally referenced files (e.g. file:fuse://conf/tasks/builder.yml ) will be automatically uploaded to the DBFS and referenced in the job definition.
• Package containing all the project code will also be automatically built and uploaded to DBFS

All these actions are to be performed when the following command is launched:

dbx deploy --job=e2e-mlops-demo-main

After running it, check out the job in Workflows tab:

When this job is getting launched, data will be first downloaded and then a new hyperopt search process will be launched. At the end of builder step, a new model version will be created in the MLflow Models Registry. This registry is available via the Models tab in Machine Learning lens:

This model version is not yet in any stage, because we haven’t yet defined any stages for it. This is something we’ll do in the next part when the model will be deployed to a serving layer.

Parameter search scalability

We all like performance graphs, aren’t we? Here is another one — training a 100 of model evaluations using various amount of worker nodes, specifically 1,2,6,10 and 14 workers configuration.

Before running the performance tests, some general considerations were taken into account to produce comparable results:

• All tests were performed on Azure Databricks Jobs cluster with Databricks ML Runtime 11.1.x-cpu-ml-scala2.12.
• All cluster nodes (both driver and workers) were pre-warmed to avoid VM startup time effects.
• Average worker node utilization was around 85–90% during the active training phase.
• Since num_estimators was also one of the parameters in XGBoost, it might happen that some tasks were heavier than others. However, most of the training results chose num_estimators in range from 80 to 100.

You could also read more about some of configuration details below:

The stats for the runs are described below:

From these values we could see that in this specific case hyperopt brings pretty good scalability in terms of parallelising the hyper-parameter search process.

Summary

In this part of the series the following topics were discussed:

• high-level architecture for MLOps process
• quickstart and project generation
• Various ML training scalability patterns
• Training process with hyperopt

The code relevant for this version could be found here (please note the 0.0.1 tag). In the upcoming part we’re going to deep dive into the serving topic. Stay tuned!

While working with Python notebooks and projects in a long-term scope, Data and ML engineers come to the idea of using best software development practices in terms of storing and versioning code artifacts.

One such practice is to organize common, widely re-used code blocks into packages, and store these packages in a reliable system that provides versioning and authentication capabilities.

In the case of an OSS project in Python, the most obvious choice would be to store your released projects directly in the main Python Package Index (PyPI). However I doubt that you would like to keep your enterprise code in a public space, so we need to organize a private package index.

Whilst there are multiple alternatives on how to organize a private package index, I prefer ready-to-use cloud solutions for that. Fortunately, Azure DevOps includes functionality for storing packages, and it’s called Azure Artifacts.

If you’re looking for the source code of the examples used in this blog post, look here.

Architecture

Here I’ve briefly outlined the solution architecture and the main components:

We start from the local repository, and on each push, we run a CI pipeline. When the time for a release comes, we trigger the release one, and it delivers the prepared package to Azure Artifacts.

On the Databricks side, we connect to the artifact repository (to properly pass auth variables we’ll set up an init script), and then we’re good to install and use our custom package in notebooks and jobs.

Writing the package

First, let’s assume that we’re writing a package that contains common Python functions which are used by many downstream teams in our organization. We’ll create a project based on standard Python structure:

.
|-- README.md
|-- azure-pipelines.yml
|-- databricks_python_packaging
|   |-- __init__.py
|   `-- utils
|       |-- __init__.py
|       `-- common.py
|-- notebooks
|   `-- usage_example.py
|-- pytest.ini
|-- scripts
|   `-- pip-config.sh
|-- setup.py
|-- tests
|   `-- unit
|       |-- conftest.py
|       `-- vertical_append_test.py
`-- unit-requirements.txt

Quick overview of the repo content:

• azure-pipelines.yml contains our CI/CD pipeline definitions
• databricks_python_packaging is a folder with the package source code
• setup.py is a python project file that describes the name, version, and other package properties.
• all other files are relevant for testing and local dependency management

Assume we currently have one function in our package:

It’s a fairly simple function that vertically appends multiple data frames into a single data frame.

Following software development best practices, we indeed shall also add tests for this function:

I wouldn’t say it’s a good test — it only covers positive cases, but this time it’s not in the focus.

Now let’s create a new project in Azure DevOps and add the Git Source to it:

In the screen above I’ve created a project databricks-python-packaging and added the GitHub repository with the source code to it.

I also created a new feed in the Artifacts section, and made it project-scoped.

The next step is to define the CI/CD pipelines. I’ll only add the relevant parts here, please review the whole file in the linked repository if interested.

CI pipeline installs dependencies, runs tests, and collects coverage metrics:

And the release pipeline does the most relevant part — builds & uploads the distributed file to the Azure Artifacts feed I created earlier:

Now the classical steps for a release:

• bump version in __init__.py file
• commit & push code to the main branch
• create & push tag:

git tag -a "vX.Y.Z" -m "Release X.Y.Z"
git push origin vX.Y.Z

As a matter of fact, bumping versions could be automated with tools such as dunamai or Versioneer, but I’ll keep this automation bit away for the sake of focus.

After pushing the tag, the release pipeline will do its job and the released package will appear in the Azure Artifacts feed:

Installing package locally

Imagine now that the developers from one of the teams would like to use this package during the local testing scenario. In my case, it’s a Macbook with Apple M1, so I needed to install .NET SDK via installer and also install dependent packages:

pip install keyring artifacts-keyring

It’s also required to add an additional package index to the pip.conf file — follow these instructions.

When all pre-requisites are done, I can download and install the package via:

pip install databricks-python-packaging

And use the function I’ve created in my own code.

Installing package in Azure Databricks

The final step is to install the package in the Azure Databricks environment. For that we’ll need to use the cluster node init scripts functionality, to configure the pip.conf file.

The simplest way to do it is the following:

• Create a new PAT token in Azure DevOps with permissions to read packages:

• Add the following secrets to the AKV in the same way:
• the above-generated token
  - Azure DevOps feed name
  - Azure DevOps organization name
  - If your package is project-scoped (as in this example), add the project name as well.

Please note that this setup expects you to have an attached AKV to the Azure Databricks workspace. Find more on how to organize such a setup here.

Security note: for simplicity and demonstration sake I’m doing the steps above manually. In a full-scale enterprise setup these steps shall be peformed either via Azure ARM or Terraform provider (or any other IaaC solution). Don’t manage your tokens manually and always rotate them.

On the Databricks side of things, we’ll need to create a new init script which will populate the pip.conf file:

Now we can configure the interactive clusters to use this script and relevant secrets (don’t forget to add the uploaded script to the cluster init scripts configuration):

After this setup, any cluster with the attached init script and provided env variables will be able to install the provided package:

%pip install databricks-python-packaging

Let’s use our newly packaged function for demonstration purposes:

Summary

Cloud platforms provide ready-to-use interfaces for software development best practices, although it requires a bit of infrastructure setup.

For the sake of focus, I’ve intentionally avoided going too deep into infra-level setup (although it’s not the hardest thing to automate). Another simplification step is to add a cluster policy that automatically propagates the env variables mapping to the cluster definitions.

The closer your data product is getting to the production, the bigger is the importance of properly collecting and analysing logs. Logs help both during debugging in-depth issues and analysing the behaviour of your application.

For general Python applications the classical choice would be to use the built-in logging library which has all the necessary components and provides very convenient interfaces for both configuring and working with the logs.

For PySpark applications, the logging configuration is a little bit more intricate, but still very controllable — it’s just done in a slightly different way, contrary to the classical Python logging.

In this blogpost I would like to describe approach to effectively create and manage log setup in PySpark applications, both in local environment and on the Databricks clusters.

If you are looking for the source code, please find it here.

A little bit of theory

Apache Spark uses log4j package under the hood. Therefore, to properly pass logging messages and unify them across multiple APIs, we’ll need to initialize the logger in a slightly different way, compared to the usual logging package for Python applications.

In a classical Python environment the logger instance is initialized in the following way:

import logging 

logger = logging.getLogger("logger_name")

However for PySpark applications, we’ll need to use the py4j bridge to properly initialize the logger:

In this code we do the following:

• Get the LogManager object via py4j bridge
• Obtain the class name with it’s module name and optional custom_prefix
• Return the initialized logger object

However, simply creating the logger object is not enough, since we also need to provide log4j.properties file. In this file we’ll define the formatting of the logging strings and different levels for packages from which we’re going to collect the logs.

Apache Spark setup

While running PySpark applications locally, you most probably are going to install the pyspark package via pip:

pip install pyspark

And in your application code you most probably are going to initialize the SparkSession object via the following block of code:

class SomeApplication:
    def __init__(self):
        self.spark = SparkSession.builder.getOrCreate()

It’s pretty simple to add to this block the logger definition, based on the above-mentioned snippet:

# located in some_package.some_module

class SampleApp(LoggerProvider):
    def __init__(self):
        self.spark = SparkSession.builder.getOrCreate()
        self.logger = self.get_logger(self.spark)

By inheriting the LoggerProvider class, we simplify the process of recognising the SampleApp module name, so in our logs it will be represented as:

some_package.some_module.SampleApp <here goes the message>

Now to use this logger object we can simply add the following:

class SampleApp(LoggerProvider):
    def __init__(self):
        self.spark = SparkSession.builder.getOrCreate()
        self.logger = self.get_logger(self.spark)
    
    def launch(self):
        self.logger.debug("some debugging message")
        self.logger.info("some info message")
        self.logger.warn("some warning message")
        self.logger.error("some error message")
        self.logger.fatal("some fatal message")

Please note the method names for the self.logger object — they’re not the same as a classical logging.Logger API. Since this logger object is based on log4j, it shall follow the log4j methods (e.g. not warning but warn and not critical but fatal).

If we now launch this application in the local environment, we’ll see that not all messages are actually getting to the output. Also, we might would like to add a bit more detailed formatting to the logs itself:

22/06/11 18:39:41 WARN SampleLoggingJob: some warning message
22/06/11 18:39:41 ERROR SampleLoggingJob: some error message
22/06/11 18:39:41 FATAL SampleLoggingJob: some fatal message

Now it’s time to introduce new file — log4j.properties . You can find a reference file here. You can easily provide a customized version of this file by using the SPARK_CONF_DIR environment variable as described here.

At first, let’s start from something basic:

If we’ll launch this with our code, we suddenly won’t see any proper logs in the output, but the formatting will be applied properly:

[spark][2022-06-13 12:29:26][WARN][org.apache.hadoop.util.NativeCodeLoader][Unable to load native-hadoop library for your platform... using builtin-java classes where applicable]
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
[spark][2022-06-13 12:29:27][WARN][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some warning message]
[spark][2022-06-13 12:29:27][ERROR][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some error message]
[spark][2022-06-13 12:29:27][FATAL][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some fatal message]

We can also notice this strange line in the log output:

Setting default log level to "WARN".

This is where thing get a little bit counterintuitive — although we might expect that the default log level for PySpark applications is taken from log4j.rootCategory=DEBUG, console , it’s actually not exactly true. We’ll need to add another configuration line when we’re running the PySpark application locally:

log4j.logger.org.apache.spark.api.python.PythonGatewayServer=DEBUG

With this line and some log formatting we’ll have the following file:

And if we launch our application now we’ll get a really rich log output (maybe even too verbose):

spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.util.component.AbstractLifeCycle][starting ServletHandler@694d6976{STOPPED}]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][Path=/[EMBEDDED:null] mapped to servlet=org.apache.spark.ui.JettyUtils$$anon$1-42fde1ed[EMBEDDED:null]]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.http.pathmap.PathMappings][Added MappedResource[pathSpec=ServletPathSpec@4e{/},resource=org.apache.spark.ui.JettyUtils$$anon$1-42fde1ed==org.apache.spark.ui.JettyUtils$$anon$1@bace2775{jsp=null,order=-1,inst=false,async=true,src=EMBEDDED:null,STOPPED}] to PathMappings[size=1]]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][filterNameMap={org.apache.spark.ui.HttpSecurityFilter-566be2b0=org.apache.spark.ui.HttpSecurityFilter-566be2b0==org.apache.spark.ui.HttpSecurityFilter@566be2b0{inst=false,async=true,src=EMBEDDED:null}}]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][pathFilters=[[/*]/[]/[INCLUDE, REQUEST, FORWARD, ERROR, ASYNC]=>org.apache.spark.ui.HttpSecurityFilter-566be2b0]]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][servletFilterMap={}]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][servletPathMap=PathMappings[size=1]]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.servlet.ServletHandler][servletNameMap={org.apache.spark.ui.JettyUtils$$anon$1-42fde1ed=org.apache.spark.ui.JettyUtils$$anon$1-42fde1ed==org.apache.spark.ui.JettyUtils$$anon$1@bace2775{jsp=null,order=-1,inst=false,async=true,src=EMBEDDED:null,STOPPED}}]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.server.handler.AbstractHandler][starting ServletHandler@694d6976{STARTING}]
[spark][2022-06-13 12:33:55][DEBUG][org.sparkproject.jetty.util.component.AbstractLifeCycle][STARTED @2251ms ServletHandler@694d6976{STARTED}]

Now it’s time to actually configure it properly, so we see only relevant messages from our application and suppress too much detailed logging information. In this example we follow the log4j.properties.template with some minor adjustments:

The logs look now way less verbose, but it seems to me that I’ve lost the debugging details from my application:

[spark][2022-06-13 12:39:06][INFO][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some info message]
[spark][2022-06-13 12:39:06][WARN][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some warning message]
[spark][2022-06-13 12:39:06][ERROR][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some error message]
[spark][2022-06-13 12:39:06][FATAL][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some fatal message]

This output is missing the debugging commands I’ve had in my application — and it makes sense since I’ve set our root and GatewayServer configuration to INFO level. Here the log4j configuration flexibility comes in handy. Since I know that the name of my package is pyspark_logging_examples — I can adjust my logging output based on it’s name by simply adding one line:

# Custom line for my Python packagelog4j.logger.pyspark_logging_examples=DEBUG

Now we’re talking — all logs are properly printed:

[spark][2022-06-13 12:45:20][DEBUG][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some debugging message]
[spark][2022-06-13 12:45:20][INFO][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some info message]
[spark][2022-06-13 12:45:20][WARN][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some warning message]
[spark][2022-06-13 12:45:20][ERROR][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some error message]
[spark][2022-06-13 12:45:20][FATAL][pyspark_logging_examples.workloads.sample_logging_job.SampleLoggingJob][some fatal message]

In the same manner you can mute (or add) more detailed logs per different components of Apache Spark — just notice the ones that are too verbose and point them to a higher level, e.g. if I think that the DAG scheduler logs are too verbose on the INFO level, I can simply cap them to WARN level:

log4j.logger.org.apache.spark.scheduler.DAGScheduler=WARN

Summarizing on this path, here are some important things to consider when driving PySpark application logs in the OSS Apache Spark environment:

• Use the Spark-based logger and not the logging-based one
• Provide your logging configurations in log4j.properties file and place it in the SPARK_CONF_DIR path
• Don’t forget to keep in sync the rootCategory and PythonGatewayServer logging properties
• You can easily customize your logging levels per different loggers (which could be your application or specific Apache Spark class or package)

Managing logging configurations in the Databricks environment

On the Databricks platform, we’ll reuse the above mentioned class called LoggerProvider to write the logs:

After running this code, in the Driver Logs tab, in the log4j output we’ll see the following lines:

22/06/13 11:20:00 INFO LoggerProvider: some info message
22/06/13 11:20:00 WARN LoggerProvider: some warning message
22/06/13 11:20:00 FATAL LoggerProvider: some fatal message

However we also would like to slightly customize the logging output (add more formatting and add the fully qualified class names to it). We can easily do this by using the cluster node initialization script as described here.

We’ll need to change the layout of the publicFile appender to the following:

log4j.appender.publicFile.layout.ConversionPattern=[spark][%p][%d{yy/MM/dd HH:mm:ss}][%c][%m]%n

And now in the log4j section we see pretty clean log output:

[spark][INFO][22/06/13 11:56:01][notebooks.__main__.LoggerProvider][some info message]
[spark][WARN][22/06/13 11:56:01][notebooks.__main__.LoggerProvider][some warning message]
[spark][FATAL][22/06/13 11:56:01][notebooks.__main__.LoggerProvider][some fatal message]

however we might have a lot of different notebooks using the same cluster at a time, so it would be great to properly prepend the path to uniquely distinguish different logs:

And now our logs provide full visibility on the path on the source notebook:

[spark][INFO][22/06/13 12:03:32][notebooks.some.path.sample-logs-collection.__main__.LoggerProvider][some info message]
[spark][WARN][22/06/13 12:03:32][notebooks.some.path.sample-logs-collection.__main__.LoggerProvider][some warning message]
[spark][FATAL][22/06/13 12:03:32][notebooks.some.path.sample-logs-collection.__main__.LoggerProvider][some fatal message]

By further customizing the log4j.properties you can suppress specific components and classes from the logs (or enable a more in-depth debugging for specific parts).

You can find all the above mentioned code pieces and more profound example in this repository.

In this post I’ve tried to cover different options for configuring log4j and using it as a main logging engine for a PySpark application both in OSS Apache Spark and Databricks context. The one topic which remained not properly covered is the setup, configuration and collection of the logs on the worker nodes.