DEV Community: Agile Developer

Yet another end-to-end streaming dashboarding example

Agile Developer — Tue, 12 May 2026 11:23:13 +0000

Introduction

In this post, we present an introductory example using Apache Pinot to ingest an Apache Kafka stream. This is an introductory post that builds upon existing Apache Pinot material from the official trainings and documentation. The purpose here is not just to rehash what is in the official docs, but a preparation for a second part. The idea, is to adapt the official examples to this end. Moreover, when I tried to run these examples, I had some extra ideas in how to better present the material. Part of the presented setup is also based on yet another Apache Pinot example in a complementary series of lectures that is written for Javascript. Our focus here is Python. Here are the two references I used

Lecture 4 (https://github.com/startreedata/learn/tree/main/pinot-advanced/04-stream-ingestion). It is a series of advanced Pinot usage from Startree. I Ported the JS example to Python.
Updated continuously Streamlit example

Another purpose of this introduction is to document my learning process so as to use it later as a reference or personal notes. Consequently the coherence of the material presented is of paramount importance.

For a formal introduction to Apache Pinot, the excellent playlists below are highly recommended.

Apache Pinot 101
Apache Pinot 201

Let's start our journey.

Booting up setup and running our first streaming session

Our setup is completely local. We will use exclusively Podman. All the executions are done on Windows 11 using Command Prompt terminals under VScodium. You might need to apply some minor changes for your environment (if any).

The docker compose file is mostly covered here . We just added a .env file for convenience.

podman compose up -d

This starts an Apache Kafka single-node cluster and an Apache Pinot cluster with one Controller, one Broker and one Server nodes. More on this later. You can visit the Apache Pinot Controller UI here.
Having started Apache Kafka and Apache Pinot we need to push some data to Apache Kafka and link Apache Pinot to Apache Kafka through a streaming table. As in both references, we will use Wikipedia page edits event stream as a data source. Every page edit on Wikipedia is recorded as a event. There are many page edits throughout the world in an ever increasing body of knowledge on Wikipedia. This happens, literally continuously and such activity can be modeled as an event source. This event is made public in the following url https://stream.wikimedia.org/v2/stream/recentchange and people can visit it with their browser and see these events. Obviously, the typical web surfer is not interested in this overwhelming, ever growing list of repetitive JSON context. It is so large that one has to resort to Data Analytics methods, so as to make sense. Moreover, this event stream is not structured in a way to convey meaning as a typical web page. On the contrary, methods of Data Engineering are necessary to capture it in a streaming table (Apache Spark terminology is used here), do whatever data transformations are necessary and then make it available to a Data Analytics system for visualizing the different aspects.
First, we need to understand the data source. The data source is delivered in what is commonly referred to as SSE format. Wikipedia, unsurprisingly has a very detailed page with documentation on this. It also lists various code snippets on how to consume it. In terms of Data Engineering,

is a web service that exposes continuous streams of structured event data. It does so over HTTP.

For a Data Engineer, a source transport format is half the story. The rest is the schema. It is available here.

In terms of software development, this means, that we need a client library. There are many, but SSE client stands out. It is also used in the Streamlit tutorial of Apache Pinot. For simplicity, we will use the Wikipedia approach.

Here is the adapted code from Wikipedia.

url = 'https://stream.wikimedia.org/v2/stream/recentchange'
headers = {"User-Agent": "advanced_pinot_tutorial"}

with EventSource(url, headers=headers) as stream:
    for event in stream:
         if event.type == 'message':
            try:
                change = json.loads(event.data)
                change['ts'] = change['timestamp'] * 1000
                del change['timestamp']

                # Kafka Place holder Code is here

            except ValueError:
                pass

From the schema what stands out for a streaming source is the timestamp

timestamp:
description: Unix timestamp (derived from rc_timestamp).
type: integer
maximum: 9007199254740991
minimum: -9007199254740991

The above conversion is to avoid a conflict with any internal timestamp function. Also we convert the Unix timestamp to milliseconds. Keep it in mind.

Now we need some code to push to an Apache Kafka topic. We use the confluent-kafka library.

First we setup our Apache Kafka connection (we implicitly assume the default 9092 port for the Apache Kafka), which is petty much self-explanatory

kafka_topic_name = "wikipedia-events"

# conf = {'bootstrap.servers': 'redpanda-0,redpanda-1,redpanda-2'}
conf = {'bootstrap.servers': 'kafka'}

kafka_admin = admin.AdminClient(conf)

kafka_admin.delete_topics([kafka_topic_name])
kafka_admin.create_topics([admin.NewTopic(kafka_topic_name, 1, 1)])

producer = Producer(conf)

and then in the Apache Kafka placeholder in the previous snippet we put the push logic

producer.poll(0)
producer.produce(kafka_topic_name, key=change["meta"]["id"], value=json.dumps(change), callback=acked)

events_processed += 1
if events_processed == 100:
    print(f"{str(datetime.datetime.now())} Flushing after {events_processed} events")
    producer.flush()
    events_processed = 0

every 100 events, we log the push of the batch. Confluent has very good documentation on how this library is used.

We pack the application a Docker image

podman build -t pinot-advanced/python-streaming-ingest ./producer-app

and then, we run it

podman run -it  --network=pinot-advanced pinot-advanced/python-streaming-ingest:latest

Now it is time to verify the Apache Kafka push is working appropriately. For convenience a consumer Python app is provided. You can start it with similar commands

podman build -t pinot-advanced/python-kafka-consumer ./consumer-app
podman run -it  --network=pinot-advanced pinot-advanced/python-kafka-consumer:latest

Everything seems to work fine.

Setting up Apache Pinot and running our first query

In order to create the streaming table, we need to tell Apache Pinot both the transport format and the schema. The schema need not be exhaustive, but include a subset of what we need. For this reason we need two files.

The schema file.

Each column in Apache Pinot has one of the following types.

Dimension
Metric
Date/Time

It is pretty obvious what the last one is used for. The first one is for filtering (used for drilling down). The second one is for aggregations. This distinction does not exist in relational databases or other Big Data solutions, and is what makes Apache Pinot a true Big Data streaming solution.

We will not need any metric fields, since we get a stream of data edits. We will do what people call distinctCounts which in reality is an aggregation, but the fields we will use are not numeric and so, they cannot go to the metric fields section. Here you are

{
  "schemaName": "wikievents",
  "dimensionFieldSpecs": [
    {
      "name": "metaJson",
      "dataType": "STRING"
    },
    {
      "name": "user",
      "dataType": "STRING"
    },
    {
      "name": "domain",
      "dataType": "STRING"
    },
    {
      "name": "topic",
      "dataType": "STRING"
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ts",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

The config file.

Next one is the table configuration and transport format. See https://github.com/fithisux/visualize-streamlit-pinot-example/blob/main/scripts/wikipedia_events_realtime_table_config.json for the details.

I will just focus on this snippet

{
    "transformConfigs": [
      {
        "columnName": "domain",
        "transformFunction": "JSONPATH(metaJson, '$.domain')"
      },
      {
        "columnName": "topic",
        "transformFunction": "JSONPATH(metaJson, '$.topic')"
      }
    ]
  },

It is necessary, so as to grab the fields from the JSON payload of the Apache Kafka message. So, fields topic and domain are computed fields, and for this reason we need explicitly expose the metaJson column.

Our first query

With the compose file and streamer app up and running we will construct our table in Apache Pinot.

podman run -it --network=pinot-advanced -v ./scripts/wikipedia_events_schema.json:/scripts/wikipedia_events_schema.json -v ./scripts/wikipedia_events_realtime_table_config.json:/scripts/wikipedia_events_realtime_table_config.json apachepinot/pinot:latest-25-ms-openjdk AddTable -schemaFile /scripts/wikipedia_events_schema.json -tableConfigFile /scripts/wikipedia_events_realtime_table_config.json -controllerHost pinot-controller -exec

We mount ./scripts on a purpose built container that will use schema and table config in order to create the table.

You can view the table by navigating to Pinot Controller locally here and run your first query

select domain, topic, user, ts from wikievents limit 10;

Here is a sample of what you should expect

Running the dashboard

Deviating from the sample Streamlit app provided by Startree, but similar in spirit we provide a Dashboard. Before delving into the code base let's clarify the business logic of the dashboard. We run a sampling query that works on a window from the sampling time, 1 minute back into the past. In this window we sample three important quantities:

The number of changes that happened
The different users that committed these changes
The different domains where this change took place.

Our dashboard will carry the current sample, and a window back in time of the 30 latest samples. For visualization we will will record the sample, and we will plot the 30 samples buffer as a visual summary. Our dashboard will be implemented with the Panel python package in a notebook. Is used VScodium for convenience. It is advised to create a virtual environment, install the dependencies there and then use it as a kernel for executing the notebook.

How is the sample obtained is just an Apache Pinot query away:

select 
   count(*) AS events1Min,
   distinctcount(user) AS users1Min,
   distinctcount(domain) AS domains1Min
from wikievents_REALTIME
where ts > ago('PT1M')
limit 1;

ago function uses ISO 8601 duration format to construct a bound for the window.

This is our main building block. To implement our sampling logic here is the relevant notebook cell

from pinotdb import connect
import pandas as pd

conn = connect(host='localhost', port=8099, path='/query/sql', scheme='http')

list_of_samples = []

def get_changes():
    query = """
        select 
                count(*) AS events1Min,
                distinctcount(user) AS users1Min,
                distinctcount(domain) AS domains1Min
        from wikievents_REALTIME
        where ts > ago('PT1M')
        limit 1;
    """

    curs = conn.cursor()

    curs.execute(query)

    temp_df = pd.DataFrame(curs, columns=[item[0] for item in curs.description])
    temp_df['sample_time'] = pd.Timestamp.now()

    list_of_samples.append(temp_df)
    if len(list_of_samples) > 30:
        list_of_samples.pop(0)

    return temp_df.to_dict('records')[0], pd.concat(list_of_samples).sort_values(by=["sample_time"])

The sample is returned as a dict, while the past buffer is concatenated to a pandas data frame. A sample execution follows

({'events1Min': 2216,
  'users1Min': 362,
  'domains1Min': 80,
  'sample_time': Timestamp('2026-05-12 12:58:12.996165')},
    events1Min  users1Min  domains1Min                sample_time
 0        2216        362           80 2026-05-12 12:58:12.996165)

The next cell sets up the reactivity of our data

# Necessary for reactive pandas
import panel as pn
import hvplot.pandas 

pn.extension()

sample_df, samples_df = get_changes()
table_changes = pn.rx(sample_df)
samples_df_rx = pn.rx(samples_df)

## Extract Data

def update_table_changes():
    sample_df, samples_df = get_changes()
    table_changes.rx.value = sample_df
    samples_df_rx.rx.value = samples_df

pn.state.add_periodic_callback(update_table_changes, period=60000)

See documentation of Panel library here. The most important statement is the last one that sets up 1 minute periodicity of updates for our feeds to plots.

The next cells create a dashboard with the absolute defaults. No effort to tinker with CSS is taken. I will not spend time on the Panel components. The documentation is very thorough. What is remarkable though, is that you can directly serve the notebook with Panel. From you activated virtual environment run

panel serve .\dashboard.ipynb

and you can navigate to the appropriate url http://localhost:5006/dashboard to visit your dashboard

Epilogue

In the above article we gave an example of an end-to-end dashboard backed by Apache Pinot streaming table. The original stream comes from an Apache Kafka topic. The stream captures the Wikipedia page edits and is customarily used for streaming tutorials. We gave a quick description of Apache Kafka and Apache Pinot setup, how to ingest the page edits and how to visualize them. RedPanda can be used instead of Apache Kafka. See the related Readme.md for the necessary, but minimal, changes. As always the code is provided. If you find something is not clear, a bug, or have any suggestion, do not hesitate to post on the comments. I hope you enjoyed it.

Your ILP solver license has expired. Now what?

Agile Developer — Mon, 04 May 2026 11:08:52 +0000

Background

A nasty surprise

Last summer while trying to deliver a feature for one of our customers, I encountered a nasty situation. The software we were developing, depended on a production grade license of Gurobi. People were on vacations except of my team and some unrelated staff, so developing the feature was in principle blocked. As I learnt due to some other situations, research stuff being participating in conferences, they could not update the license. These are the people who had the final saying. Still the situation for me was very uncomfortable, because this feature would be delayed a lot. Months before I had cautioned that the sole dependency on a closed source solution was a bad practice when there were free open source solutions like HiGHS. Gurobi is the leading player in the field with a very performant product that offers many conveniences. Actually, much more performant than the open source solutions in our case. But license disruptions could happen and users of the feature would be in a difficult situation.
In summary the feature amounted to the following workflow. Users could parametrize a process in a Web GUI. These parameters are translated to an ILP (Integer Linear Programming) problem which subsequently is solved and results are returned back to the WebGUI. We followed the standard approach of sending these parameters as a REST payload to a server. The server would do the translation to the ILP. Having also done the solution, the results are sent back.

You can get a taste of that here

The plan

Having some time available I decided to evaluate the possibility of providing an alternate implementation of the solution part instead of mocking it. It was important since performance considerations were also in scope. The first attempt bombed because the code was not clean. It was written by researchers after all. I was lucky enough to have some of their notebooks with outputs for comparison. So, given this opportunity, I went ahead to clean up their code considerably (and fix a number of serious bugs, yay!!!). This post focuses on the bringing up of the alternative and not the other parts of the feature that were equally important. But first let's outline the plan of attack I decided upon. We are talking about a Python code base.

Cleanup the code so that the ILP problem is clear. Given the previous attempt of a colleague who worked on the cleanup before, I was able to further the cleanup, attach types, and make sense of the code. I will not get into more details, but it was not very pleasant.
Given the Gurobi code, and the fact that there is an interchange format for ILP problems, called MPS, the workaround here was to serialize the Gurobi formulation to an MPS file, load it and solve the ILP with HiGHS. It involved some work, mostly writing a bunch of adapters and understanding how HiGHS works. This was the path of least resistance and worked fine. Acknowledging the bottleneck of moving the huge MPS file across the network instead of the way smaller set of parameters, as the original plan was, I hid the file generation within the computation server.
While not having the best solution, I was more confident. The whole feature was progressing after all. I decided to give a shot in the re-implementation with HiGHS which would bring me in parity with the original plan. This would eliminate the serialization/deserialization of a big file. It was now easier than I anticipated.

Obviously I will not be able to share the code, but I will use a toy example to highlight the principles.

Highlights of the porting

Introduction

As a toy example I will use the famous "assignment problem". It is a very common and simple ILP problem, that pales in comparison to the ILP problem of the customer. However it is enough to highlight the main issues. I use this excellent reference. It is a good set of lectures for solving ILP problems. You can try to replicate what is presented here for the other problems.
The typical assignment problem amounts to assigning M people to N jobs with every possible assignment, say job -> person incurring a cost of C(job, person). The task is to find the minimum cost assignment. The constraints are:

Every job must be assigned exactly one person
Persons can be assigned to at most one job.

Obviously M should be at least N to cover all the jobs and M should be at most N to not leave people out. Our plan here is to solve this in three ways:

Gurobi (Model in Gurobi and solve in Gurobi)
Pseudo Gurobi (Model in Gurobi solve in HiGHS)
HiGHS (Model in HiGHS and solve in HiGHS)

Code is here.

Gurobi approach

First of all we will use named binary variables to refer to our potential assignments. If they take the value 1 after a solution, these assignments have been realized.

import gurobipy as gp
from gurobipy import GRB

env = gp.Env()
model = gp.Model(env=env)

x = {}
for job_index in range(0, Njobs):
    for worker_index in range(0, Njobs):
        var_name = f"x_{job_index}_{worker_index}"
        x[(job_index, worker_index)] = model.addVar(vtype=GRB.BINARY, name=var_name)

Now we need to have some assignment costs as we said previously.

import random
from typing import Dict, Tuple

random.seed(0)

cost: Dict[Tuple[int, int], float] = {}

for job_index in range(0, Njobs):
    for worker_index in range(0, Njobs):
         cost[job_index, worker_index] = random.randint(2 , 4) * 0.5

We selected random weights (fixing the random process by the seed for reproducibility) because if all the costs were the same an assignment of the form i -> i for every i, would be enough.

Now it is time for the constraints and the objective which model exactly what we said in the previous subsection

# all jobs must have an assignement
for job_index in range(0, Njobs):
   model.addConstr(gp.quicksum(x[job_index, worker_index] for worker_index in range(0, Njobs)) == 1)


# all workers must have at least an assignement
for worker_index in range(0, Njobs):
   model.addConstr(gp.quicksum(x[job_index, worker_index] for job_index in range(0, Njobs)) <= 1)


# objective function
objective = gp.quicksum(cost[job_index, worker_index] * x[job_index, worker_index] for job_index in range(0, Njobs) for worker_index in range(0, Njobs))

model.setObjective(objective, GRB.MINIMIZE)

This covers the first part, namely, the modeling of our problem. The second and last part is the solution.

It is enough to invoke the process

model.Params.timeLimit = 200.0 # seconds
model.Params.LogToConsole = 1
model.Params.IntegralityFocus=1

model.optimize()

The rest of the code is just for displaying the solution. Not a big deal. What is the deal breaker is the following notification from the library

Restricted license - for non-production use only - expires 2027-11-29

This means two things. The first is that we are working on borrowed time. The second has to do with the size of the problem we solve. If we set Njobs = 100 we are greeted with a crash.

GurobiError: Model too large for size-limited license; visit https://gurobi.com/unrestricted for more information

This work is in the gurobipy_formulation.ipynb

Pseudo-Gurobi and HiGHS approaches

In my case I was greeted with the "Unauthenticated" error because the license had expired and the exact error when I tried to run without the license. But not all is, lost. The solution, which is the selling point of Gurobi, is not working. However, the modelling part works perfectly. Armed with this knowledge I decided to follow the hybrid method. Model in Gurobi, solve in HiGHS. It is true that the documentation takes a bit to get used but I had to do only 2 changes. The first and more important is to swap the solution process. Because of the interoperability (an underappreciated concept haunting the Software Engineering business) it was painless. More specifically we swap this

model.Params.timeLimit = 200.0 # seconds
model.Params.LogToConsole = 1
model.Params.IntegralityFocus=1

model.optimize()

with this

import highspy

h: highspy.Highs = highspy.Highs()

model.write('mymodel.mps')
status = h.readModel('mymodel.mps')
print('Reading model file mymodel.mps returns a status of ', status)

h.setOptionValue("time_limit", 200)
h.solve()
print('Model has status ', h.getModelStatus())

Simple as that. The second change which understandably is HiGHS specific has to do with the pretty printing of the solutions.

This work is in pseudogurobipy_formulation.ipynb notebook.

Now for the pure HiGHS approach we replace the model instantiation. In other words we swap

import gurobipy as gp
from gurobipy import GRB

env = gp.Env()
model = gp.Model(env=env)

with this

import highspy
h = highspy.Highs()

Keep in mind, that this is the first part of the hybrid solution approach. Now we do not need the MPS file anymore. The solution process is simply a swap of this

model.Params.timeLimit = 200.0 # seconds
model.Params.LogToConsole = 1
model.Params.IntegralityFocus=1

model.optimize()

with this

h.setOptionValue("time_limit", 200)
h.solve()
print('Model has status ', h.getModelStatus())

What changes slightly is in the modeling. We have to define a utility function quicksum to mimic and replace the provided utility function gp.quicksum.

The second change has to do with how we instantiate a variable. We swap

 x[job_index, worker_index] = model.addVar(vtype=GRB.BINARY, name=var_name)

with this

x[job_index, worker_index] = h.addBinary(name=var_name)

As you can see there is an easy swapping. what was not easy was to cleanup and debug the modelling process which is not straightforward at all.

This work is in the highspy_formulation.ipynb notebook.

Epilogue

We show how a problem that seemed insurmountable had two solutions. Not ideal, but still solutions. While the license of a production ready commercial ILP solver expired, we can still employ slower processing so as to keep the business moving. Not only that, I had to carefully review my options and cleanup the code base to make it amenable for applying the workaround. In the process the code became cleaner, bug free and I re-evaluated some modelling approaches (I did not mention it previously). They were approved by the researchers. The end result narrowed quite a bit the memory and processing gap between the Gurobi and HiGHS approaches. Since then, we had renewed the license and the feature is delivered. This time, we are prepared for a possible outage. I hope you enjoyed the article.

As always the code is provided. Feel free to open an issue if you see something wrong or add a comment.