In this episode of the Data Engineering Central Podcast, I’m joined once again by Kevin Martin, who spent nearly 13 years working at Apple before deciding he’d had enough of the corporate grind. When we last spoke, Kevin had just left Apple and was preparing to spend much of 2026 backpacking some of America’s most beautiful wilderness.

Unmapped

Left Apple after ~13 years to backpack 1,000 miles. What happens when you stop optimizing and let the trail shape what’s next?

By Kevin

Well, Kevin is back, and he’s got some stories to tell.

Since our last conversation, Kevin has completed hundreds of miles of backpacking, including the Benton MacKaye Trail and the legendary John Muir Trail. Along the way, he spent a terrifying night with multiple black bears circling his tent in the Smoky Mountains, experienced the incredible beauty of the Sierra Nevada, and learned some unexpected lessons about himself.

One of the most interesting things Kevin discovered was that leaving the corporate world doesn’t necessarily mean leaving behind the mindset that made you successful there. After years of working in a high-performance environment at Apple, he found himself approaching retirement and backpacking with the same intensity, planning, and expectations that he’d brought to his career.

• We dig into that realization and what it means to actually enjoy the financial freedom you’ve worked so hard to achieve.

Of course, we also wander into plenty of other topics, including the changing landscape of Big Tech, AI-driven layoffs, whether anyone’s job is really safe anymore, and the importance of building financial independence in an increasingly uncertain world.

• We talk about lifestyle inflation, the FIRE movement, the psychology of money, and why doing difficult things in the outdoors can completely change your perspective on everyday problems.

Perhaps my favorite part of the conversation is our discussion about happiness and the tendency to constantly chase the next big milestone. Whether it’s a promotion, a financial goal, retirement, or climbing a mountain, there’s always something else waiting on the horizon.

It’s a fun, thoughtful conversation that goes well beyond technology and money. If you work in tech, dream about financial independence, love the outdoors, or simply find yourself wondering what life might look like outside the traditional career path, I think you’ll enjoy this one.

Grab a cup of coffee and join us!

As the world changes, especially the data world … with the advent of Agentic capabilities … like it or not … your data job is going to change; you either get on board and go along for the ride, or put your head in the sand and leave the rest to chance. I encourage you to be skeptical, but honest about the positive benefits of LLMs/AI in your day-to-day work.

is a creator I watch closely and of course he’s been on the the podcast before. We need to get him to come back.

Recently, Andreas published a YouTube video that is probably one of the best, honest, and simple examples of NOT just more AI hype … but how Agentic capabilities can become part of our everyday Data Platform in ways that can make our lives easier, better, and provide real ROI by breaking down barriers and solving problems in novel ways.

In this video, Andreas walks through building a Data Quality Assistant, connected to Databricks and Google Drive, and using MPC and SDKs from Zapier. What stood out to me is that my data analysis workflows have changed drastically over the last few years. It almost happened imperceptibly.

Of course, I use Databricks daily, and I spend that time researching, answering questions, inspecting data, etc. In the past, of course, this was very time-consuming. Sure, I had the product and business knowledge, combined with the technical skills of SQL and Spark … I could, but not without a lot of work, build a needed dashboard or chart, research a topic, and find proof in the data. These days my workflow has changed.

Subscribe now

Now I use English + my domain knowledge to instruct Agents and research the data to find the needed answers, or to explore various possible avenues and results … and it can do it quickly, find answers I wasn’t looking for … and generally take *most of the technical burden off myself … leaving me in the role of researcher … with an assistant.

If you are still skeptical about AI and LLMs, especially in a data context … I highly recommend you watch this 1-hour+ free video from Andreas.

Instead of your typical “Hello World,” or marketing drivel that takes up most of our feeds … this video is a refreshing approach to showing you generally what is possible in a simple and straightforward way. It’s not over the top; it’s just saying … “Hey, this is the sort of thing that is possible now.”

• At the end of the day, the particular tools are sorta irrelevant in some sense.

What you shouldn’t do is stick your head in the sand, decry the AI slop being produced … and ignore new technology. The technology and the data community will move on with or without you. It’s unfortunately brutal. Ok, if you are worried about AI slop, then you can, as usual, ensure the tools are doing things correctly and producing accurate results. Just like you did when you were the one producing the code or writing the SQL.

Imagine yourself as the Senior+ level engineer who has a junior working for you … just that it’s a digital agent and not flesh and blood. But you can approach it much the same way.

If you haven’t followed his work, he’s got a ton of free content and courses; we all still need to be learning in this agentic world!

It’s been hard to ignore all the 2.0 chitter-chatter of late. I could stop doomscrolling LinkedIn looking for a human-written data take, but then I would have missed both the Polars and DuckDB announcements about their (supposed) big and exciting 2.0 releases. I’ve used both tools on and off, both personally and in production settings for a long time. I started out with Polars, even using it with Unity Catalog and Databricks long before it was cool. Of late (the last few years), I’ve since changed my allegiance from the Dark Lord to the Lord of Light … aka DuckDB. I even went as far as to rip Polars out of a number of production jobs and replace it with DuckDB because of stability issues.

The best way to support this Substack is to click the links below and checkout todays sponsor, LakeSentry! Sponsors like this are what make this content possible!

Anywho, I’m sure you’re tired of me beating that dead horse.

Today I would like to make the teeth grind and gnash in as many inner circles as possible. To me, no golden calf is holy or off limits. The plan is simple, like me.

• Review changes and improvements called out in 2.0 for both DuckDB and Polars.

• Run my own hairbrained benchmark against datasets in Iceberg on S3 with lots of aggregation and window functions.

  • Of course, using undersized compute with oversized data

We shall throw off the shackles of those squirrely little marketers and their TCP benchmarks that have been buggered with. Who else is going to keep them all honest? Both of these tools, Polars and DuckDB, are used by legions of data folk for all sorts of work. It’s interesting to examine these big release changes, like 2.0, to see exactly what they are working on and what direction they are going.

It can tell us a lot about what the future holds and where smart people are placing their bets. All sorts of learnings for data professionals can be found. If we can identify where the winds of change are blowing these tools and their new features- we can get ahead of the pack, pick up and learn new skills now, and beat the rest of the lemmings to the punch. ex. learn Apache Arrow.

Polars 2.0

Ok, so let’s get after it. I saw a few LinkedIn posts about the Polars 2.0 release, pretty much what I expected to see after having worked with Polars alot over the last many years. Again, I like to see what the powers that be say about any release, as those are the topics and direction they deem important and set the direction for future changes, most likely. So what can we infer from their posts?

• Streaming engine default (this has been in the works for a while)

  • Faster and doesn’t cause OOM

• SQL as a first-class citizen (kinda funny for a dataframe tool)

  • Necessary to go far in the data world, chasing DuckDB

• Spill to disk support

  • DuckDB has been good at this; Polars was behind in the past.

• Classic “we are fast” benchmarks.

  • I’m not sure anyone doubts that Polars is fast; if they do, they need their head checked.

The whole streaming engine change is a good one; it’s been available for a long time via Lazy frames and collect(streaming) but has not been the default mode. What we would have known as eagar Polars before pulls everything into memory. With streaming a person can work on datasets larger than memory without worrying about OOM.

So this is eagar Polars …

import polars as pl

df = pl.read_csv("data.csv")
df = df.with_columns(
    (pl.col("price") * 1.10).alias("new_price")
)
df.write_parquet("output.parquet")

… and before 2.0 we could get Lazy and Streaming.

import polars as pl

df = pl.scan_csv("data.csv")
df = df.with_columns(
    (pl.col("price") * 1.10).alias("new_price")
)
df.collect(streaming=True).write_parquet("output.parquet")

Now this is the default with Lazy and collect() , this is good for a lot of data folk who use Polars, maybe switch from something like Pandas, and don’t really take the time to understand the features of the tool, and that lazy and eagar is even a thing.

The next change is probably one that should have been done so long ago. It’s always been a major issue for me when testing and using Polars on very large datasets on a single-node commodity machine. They were behind DuckDB on this feature, and now have finally caught up. Spill to Disk.

This feature is a big deal for Polars; it should go a long way toward making it more stable in large production environments where data pushes Polars to the max.

Share

Kinda funny, SQL as a first-class citizen in Polars.

Next, I found this so funny … just because. I mean, it’s not Polars fault that the data community is married to SQL and will never let it go. Polars is now the de facto DataFrame tool to replace old Pandas. But you just can’t have SQL support. Especially with AI writing all the code these days, data analysts and the like are more likely to know and learn SQL first before they get into DataFrame operations.

This is a very smart move by Polars.

To use SQL in Polars, you have two options: SQLContext or .sql on an existing DataFrame.

import polars as pl

df = pl.scan_csv("sales.csv")
ctx = pl.SQLContext(sales=df)
result = ctx.execute("""
    SELECT
        customer_id,
        SUM(amount) AS total_spent,
        COUNT(*) AS orders
    FROM sales
    WHERE amount > 100
    GROUP BY customer_id
    HAVING COUNT(*) > 5
    ORDER BY total_spent DESC
""")
result.sink_parquet("output.parquet")

… or

import polars as pl

sales = pl.DataFrame({
    "customer_id": [1, 1, 2, 2, 3],
    "amount": [100, 200, 300, 400, 500]
})
result = pl.sql("""
    SELECT
        customer_id,
        SUM(amount) AS total_spent
    FROM sales
    GROUP BY customer_id
    ORDER BY total_spent DESC
""")
result.collect()

Also, Polars, being a Rust-based tool in the backend, they also moving towards having a “strict” tool where they fail fast for obvious errors. Example, collect_schema() will “resolves types and catches schema-level mismatches without materializing any data” for example.

In summary, it appears Polars 2.0 from my point of view is just about hardening the core of how the tool works and performs. This is work that has been sorely needed. Sure we like fast tools, but we need reliable tools. Default streaming and out of core spilling to disk, along with first class SQL support are exactly the kind of things that will take Polars to the next level.

Polars has been the premier single-core dataframe tool for years; these changes solidify its spot in the data processing pantheon and will only lead to more adoption and larger use-cases.

DuckDB 2.0

Ok, so now we get to flip the coin to the other side and see what is coming forth in the tool that has taken the data world by storm, even going so far as to get DuckDB Labs acquired by AWS, which is quite amazing. I’m looking forward to seeing what is in this new release, if it’s just the same old stuff, or something new and exciting … giving us a taste of what the future might hold.

Honestly, it’s hard for me to imagine what they could possibly be adding to DuckDB. The extensions are already complete and all-encompassing; the speed at which DuckDB can process data is already top-notch. They are also known to be very stable and very developer-friendly; what else can they really do?

Ok, let’s just see what THEY quote as the top changes in this Cinnamon Teal release. This is important because it tells us what they (the project) care about, and the direction they are taking the tool.

Ok, so I’m not too surprised here with the initial list, and what they put on the FRONT of the list. Quack protocol. Historically, DuckDB has been an embedded database, an in-process tool to work on data. You run it inside Python, Rust, or another application or data pipeline. It’s been that perfect ad hoc query tool, and has been slowly working its way into being the next SQLite-type tool: bulletproof, fast, easy to use and install.

DuckDB 2.0 expands this model with the Quack protocol, which allows DuckDB instances to communicate over a network.

DuckDB is clearly moving toward a larger audience with features that let it become more than a single-node, in-process database … it works great for that, the best … but it’s obvious where they think the future lies. This probably has something to do with AI and Agents, but also AWS and going bigger. The release notes really focus on this new idea they are introducing to the DuckDB community … DuckDB as a Server.

Wow, this is obviously worthy of a number of blog posts in itself, which I will surely do. I like this idea of DuckDB as a server. We will return to this in future blogs.

Next, they did something which I think is core and fundemental to the way we deal with data today.

Where is all our data stored these days? The Lakehouse, of course … on cloud storage systems like S3. Because of this, when it comes to queries and data pipelines … anything a tool can do to improve network reads and how that data is handled across the network and interacts with the rest of the pipeline … is a huge win. This is why DuckDB released better “asynchronous I/O for reading Parquet and CSV files.”

This only bodes well for DuckDB, an already fast tool. Next, they released full support for VARIANT data types. This is no surprise; in the age of the Agent … we have seen all sorts of data types like this come into play in the Lakehouse formats. The days when tabular data reigned supreme might not be dead, but times are changing, and first-class support for this ragged and semi-structured sort of data is a must.

DuckDB is just covering all its bases.

Then it appears that DuckDB is still pulled into that tractor beam of SQL Server … even though they should have fought it and said no because it’s a bad idea … those nasty TRIGGERS have been added into DuckDB 2.0. Hey, I get it, they can be useful, and all the SQL hobbits probably have been licking their lips for this crap. I’m just old, grumpy, and scarred from by days of yore’ when SQL Server was the Data Warehouse. Stupid triggers.

Well, DuckDB has fallen prey to old evils; you can now TRIGGER your way into oblivion.

Benchmarking DuckDB 2.0 vs Polars 2.0

Look, I know this is what you little beggars have been waiting for this whole time. You know you like it; I know you do. Little benchmarking of our own to get those angry zealots frothing at the mouth and howling at the moon. Don’t say I didn’t warn you. We will see what we will see. No slippery TCP benchmarks here, just real life stuff.

What I will do is spin up a whopping 2GB node on Linode/Akamai (highly suggest if you’ve never used), then we will put a bunch of CSV data into S3 (a lot more than 2GB) and come up with some query to run that has to do a ton of aggregation. We will do this with both Polars 2.0 and DuckDB 2.0. I don’t really care if you finish faster; they are both fast tools … I just want to see if both hold up to their 2.0 claims. In DuckDB’s case, efficient S3 network I/O (they already spill to disk, etc.), and in Polars case … their default streaming and spill-to-disk ability.

We will use the Backblaze hard drive dataset, about 23 GB of CSV files on S3 … see if we can’t choke out Polars or DuckDB.

Now I’m not going to give either tool any help at all. I will install them on this small Linux box with uv and then just run the scripts; what happens is what happens. If one, or either of them fails, I will keep sizing up the Linode by 2GB a bump, and see at what point it can finish.

182 CSV files should give ‘em a run for the money.

Ok, over to the code. The data itself is just a bunch of hard drive failure indicators from a bunch of data centers.

Just a few command on our Linux box should do the trick. Python 3.12 is good enough eh. Then we get uv onboard.

root@localhost:~# python3 --version
Python 3.12.3
>> curl -LsSf https://astral.sh/uv/install.sh | sh
>> uv init bigtest
>> uv add duckdb==2.0 polars==2.0 boto3 fsspec s3fs psutil

Easy peasy lemon squeezy. Now for the code I hope can at least choke one of the tools, put them hobbits back in their place. Nothing like 23GB of data on a 2GB machine over the network! We shall see, we shall see! I’m going to be amazed if these tools can get it done; the query isn’t exactly easy.

Subscribe now

DuckDB 2.0 first!

import duckdb
import time
import psutil
import os

SOURCE = "s3://confessions-of-a-data-guy/testing-data/*.csv"

OUTPUT = (
    "s3://confessions-of-a-data-guy/"
    "benchmark-results/duckdb/results.csv"
)

start = time.perf_counter()

con = duckdb.connect()

con.execute("INSTALL httpfs;")
con.execute("LOAD httpfs;")

con.execute("""
    CREATE OR REPLACE SECRET benchmark_s3 (
        TYPE s3,
        PROVIDER credential_chain
    );
""")

con.execute("SET temp_directory = '/tmp/duckdb_benchmark';")

query = f"""
COPY (
    WITH drive_data AS (
        SELECT
            CAST(date AS DATE) AS date,
            serial_number,
            model,
            datacenter,
            CAST(failure AS INTEGER) AS failure,
            TRY_CAST(smart_9_raw AS DOUBLE) AS power_on_hours,
            TRY_CAST(smart_5_raw AS BIGINT) AS reallocated_sectors,
            TRY_CAST(smart_197_raw AS BIGINT) AS pending_sectors

        FROM read_csv(
            '{SOURCE}',
            header = true,
            all_varchar = true,
            union_by_name = true
        )

        WHERE serial_number IS NOT NULL
          AND model IS NOT NULL
          AND datacenter IS NOT NULL
    ),

    windowed AS (
        SELECT
            *,
            LAG(power_on_hours) OVER (
                PARTITION BY serial_number
                ORDER BY date
            ) AS previous_power_on_hours

        FROM drive_data
    ),

    analytics AS (
        SELECT
            date,
            datacenter,
            model,
            COUNT(*) AS drive_count,
            SUM(failure) AS failure_count,
            AVG(power_on_hours) AS avg_power_on_hours,
            AVG(
                power_on_hours - previous_power_on_hours
            ) AS avg_daily_power_on_change,
            SUM(
                CASE
                    WHEN reallocated_sectors > 0 THEN 1
                    ELSE 0
                END
            ) AS drives_with_reallocated_sectors,

            SUM(
                CASE
                    WHEN pending_sectors > 0 THEN 1
                    ELSE 0
                END
            ) AS drives_with_pending_sectors
        FROM windowed
        GROUP BY
            date,
            datacenter,
            model
    )

    SELECT *
    FROM analytics

    ORDER BY
        date,
        datacenter,
        model

) TO '{OUTPUT}' (
    FORMAT CSV,
    HEADER TRUE
);
"""

con.execute(query)

elapsed = time.perf_counter() - start

print(f"DuckDB 2.0 elapsed: {elapsed:.2f} seconds")
print(f"Output: {OUTPUT}")

con.close()

Any guesses?

uv run benchmark_duckdb.py

Well, it’s working hard according to Linode! Good to see that Disk in use, means we’ve got spill going on!

Hot-diggy-dog! DuckDB 2.0 pulled through! I had my doubts cooking dinner for the family and keeping one eye pointed at the terminal … I was worried for a minute or two!

DuckDB 2.0 elapsed: 837.32 seconds
Output: s3://confessions-of-a-data-guy/benchmark-results/duckdb/results.csv

That is almost 14 minutes, but dang, starved for memory and CPU, to be able to process that much data over the network and then run the analytics! What a tool. Dang.

Polars 2.0 second!

Ok, so here it goes. Got any guesses? I do. But I’m also cautiously optimistic that the new spill-to-disk features will finally throw those nasty OOM errors to the wind! I’ve been waiting for this moment since I started using Polars so many years ago!

import polars as pl
import time

SOURCE = "s3://confessions-of-a-data-guy/testing-data/*.csv"

OUTPUT = (
    "s3://confessions-of-a-data-guy/"
    "benchmark-results/polars_lazy/results.csv"
)

start = time.perf_counter()

# Lazy S3 CSV scan
lf = pl.scan_csv(
    SOURCE,
    infer_schema=False,
    storage_options={
        "aws_region": "us-east-1"
    }
)

lf = (
    lf.select(
        pl.col("date").str.to_date().alias("date"),
        pl.col("serial_number"),
        pl.col("model"),
        pl.col("datacenter"),
        pl.col("failure")
          .cast(pl.Int32, strict=False)
          .alias("failure"),
        pl.col("smart_9_raw")
          .cast(pl.Float64, strict=False)
          .alias("power_on_hours"),
        pl.col("smart_5_raw")
          .cast(pl.Int64, strict=False)
          .alias("reallocated_sectors"),
        pl.col("smart_197_raw")
          .cast(pl.Int64, strict=False)
          .alias("pending_sectors")
    )
    .filter(
        pl.col("serial_number").is_not_null()
        & pl.col("model").is_not_null()
        & pl.col("datacenter").is_not_null()
    )
)

lf = (
    lf.sort(["serial_number", "date"])
    .with_columns(
        pl.col("power_on_hours")
          .shift(1)
          .over("serial_number")
          .alias("previous_power_on_hours")
    )
)

lf = (
    lf.group_by(
        ["date", "datacenter", "model"]
    )
    .agg(
        pl.len().alias("drive_count"),
        pl.col("failure")
          .sum()
          .alias("failure_count"),
        pl.col("power_on_hours")
          .mean()
          .alias("avg_power_on_hours"),
        (
            pl.col("power_on_hours")
            - pl.col("previous_power_on_hours")
        )
        .mean()
        .alias("avg_daily_power_on_change"),
        (
            pl.col("reallocated_sectors") > 0
        )
        .sum()
        .alias("drives_with_reallocated_sectors"),
        (
            pl.col("pending_sectors") > 0
        )
        .sum()
        .alias("drives_with_pending_sectors")
    )
    .sort(["date", "datacenter", "model"])
)

lf.sink_csv(
    OUTPUT,
    engine="streaming"
)

elapsed = time.perf_counter() - start

print(f"Polars 2.0 LazyFrame elapsed: {elapsed:.2f} seconds")
print(f"Output: {OUTPUT}")

Man, we are asking alot of those two work-horse engines today, aren’t we? It just shows how far we have come in the data world over the decades. I remember the days when your only options were Pandas, Spark, or vanilla Python with some NumPy. Or, more usually, push the data to some RDBMS via ODBC and let that engine take over. Those were the days.

Ahh … I figured! Dang Polars 2.0 … you failed me!

root@localhost:~/bigtest# uv run benchmark_polars_lazy.py 

root@localhost:~/bigtest# 

OOM most likely. Bugger all, I seriously had my hopes up for this. Let’s do another Linode and give ‘er some more juice … see if it can get the job done. We will take it up to 4GB and see if that is enough for Polars not to choke.

Ok, so now Polars has an unfair advantage over DuckDB 2.0, twice the memory and CPU.

Son of a weasel! Still can’t do it!

root@localhost:~/bigtest# uv run benchmark_polars_lazy.py 
root@localhost:~/bigtest# 

Blasted Gandalf’s beard, now let’s go up to 8GB and see if Polars 2.0 can come through for us at all.

Dang, not even that was enough.

root@localhost:~/bigtest# uv run benchmark_polars_lazy.py 
root@localhost:~/bigtest# 

Clearly, something about the query means Polars can’t stream and spill to disk. Y’all have no idea how disappointed I am. Polars has been doing this for years … leaving me in a lurch. Dang it! Heck, that spill-to-disk had me so excited.

Well, this explains it … reading the release notes more closely… Polars doesn’t support spill-to-disk on any groupBy. Welp, there goes 90% of data users who would use this functionality; back to DuckDB we go!

I personally can’t think of a more important spill-to-disk feature than a groupBy , but whatever.

DuckDB 2.0 vs Polars 2.0 in review.

Well, we had fun, didn’t we? For some reason, this stuff never seems to disappoint; always a good time. I’m pretty pleased overall when I look at where these tools are generally going in the future. It’s important to read between the lines to figure out where the projects are headed and what they deem as important. It’s interesting to watch Polars try to catch up to DuckDB in terms of being able to work on large datasets that need to spill to disk and stream to finish on a single-core machine.

It seems Polars is trying to become more stable and able to have that reputation, like DuckDB, that it can eat anything and not break. Clearly Polars has some work to do yet there. Hey, at least that got the first-class SQL going for them now!

DuckDB 2.0 is very interesting; they’ve been chasing this Client/Server architecture for a little bit, and it’s super interesting to see the Duck transition to this sort of architecture. They have conquered every other data challenge and become practically a household data name. It makes one wonder, with AWS involved, where it will go in the future.

I’m super glad we took the time to go through both released AND throw some work at em. There is plenty to still learn and play with these days; no need to think that AI is taking over your job. Being creative and staying up-to-date with what is changing will never lead you astray!

UPDATE: This isn’t DuckDB 2.0 vs Polars 2.1

Of course, a week after I did my testing … Polars 2.1 was released, and I saw a little tidbit: “spill to disk for group-by and window functions.” Well, it isn’t really fair for me to compare Polars 2.1 to DuckDB 2.0, but I was so darn unhappy that Polars couldn’t operate on the dataset no matter how much RAM I gave it that I had no choice but to test it again.

Back to ye’ ole’ 2GB Lindoe machine.

>> curl -LsSf https://astral.sh/uv/install.sh | sh
>> uv init bigtest
>> uv add polars==2.1 boto3 fsspec s3fs psutil
...
>> root@localhost:~/bigtest# uv run bugger_all.py 

Well, Polars didn’t lie about 2.1 feature upgrades!

Polars 2.0 LazyFrame elapsed: 438.47 seconds
Output: s3://confessions-of-a-data-guy/benchmark-results/polars_lazy/results.csv

I hate to say it, but it’s twice as fast as DuckDB!

Firstly, I must tell you that I have mixed feelings about this; maybe it’s my crusty old age coming out to haunt me. We surely must accept that AI is here to stay, and it’s amazing the places it manages to wiggle and squirm its way into. That being said, it’s not like we could actually keep a #blessed tool like Apache Airflow clean and free from agentic sugar that keeps getting sprinkled into every possible corner. But here we are, and now it’s official; Apache Airflow supports LLM calls and Agentic actions. So, since the data boat has landed yet once again on this LLM shore, I will do my best to bring you all up to speed on the new LLM operators that now have first-class support in Airflow.

Today, I just want to do nothing more than introduce this topic. I’m sure most of you use Apache Airflow a fair bit, and it’s good to keep up to speed on the happenings, if for no other reason than to look smarter than our coworkers. Honestly, I’ve used Airflow for so long that when upgrading AWS MWAA to a 3.0+ version and dealing with the new UI layout, it reminded me how set in our ways and comfortable we get. Comfortable is not good. It’s just an important reminder that in all things tech, we should keep our ears to the ground and make sure we know which way the cattle are moving.

Exploring the new AI Operators and Task Decorators in Airflow.

I’m fairly certain I don’t need to cover this, and I’m just an everyday Airflow user, not an expert. I try to use it only for pipeline orchestration and monitoring, leaving data processing to other tools. For that reason, I’ve used it for a decade but also tried to avoid learning too much about it … other than that, I hate dynamic DAGs. Dynamic Tasks… fine; dynamic DAGs? Straight to jail.

Anywho, I’m probably only going to cover two things in this post about the addition of AI to Apache Airflow.

• Operators

  • 6 new AI-specific operators

• Tasks

  • 6 new AI-specific tasks

We will probably use that Airflow GOAT Astro CLI to spin up a local environment and maybe play with a few of the Operators and Tasks, build a little something something, and see what’s cracking. Might be boring, might not. At least we will know our stuff when that CTO comes sniffing around asking about how we are integrating AI into our pipelines.

When we are reviewing these new Tasks and Operators … remember to think about deterministic vs non-deterministic workflows. Meaning, when you put the same input in, you get the same result every time. With other tasks, you could put something in and get who knows what out the other end. This is a small but important concept when you start diving into AI inside workflows. Read the article above for more detail.

Let’s get back to our new Airflow Operators and Tasks. Let’s go through a few of them and try them out, see what we see.

Also, I’m going to use the Astronomer CLI they provide, probably the best way to run Airflow, especially locally.

Read more

I was recently chatting with of Early Signal fame, and he said two words that caught my attention before they barely escaped his lips … DuckDB and ArrowFlight … that’s all it took, I was hooked. If there are two weaknesses I have, it would be those two tools; put them together, and it should be illegal! I find it absolutely amazing we live in the age where we have raged against the machine (like Spark), and embraced these fast and lightweight tools … to build other tools!

Anyways, this has led me to be able to partner with GizmoData, which I am excited to introduce to you, and play around with myself.

Everyone else is crying into their socks and moaning about the end of all things with AI doing everything, not me and you! Heck, this is a great time to be alive and working in data! GizmoData proves there are still very smart people out there building amazingly fun tools on top of open-source tooling like Arrow and DuckDB. For ages, we didn’t have anything to pick from when it came to data tooling. I grew up and cut my teeth in an era when you chose between Pandas and Spark. That was it. One or the other.

Concepts first … then we play.

As much as I want to simply start pumping data through the system and seeing what happens, you know I love new tools; let’s get our feet under us first and just take a glance at what we are dealing with. First, things first … what is GizmoData?

The key concepts and capabilities of the open-source version …

• DuckDB and SQLite execution backends.

• Apache Arrow Flight SQL for remote query execution and columnar result transport.

• Authentication, TLS, JWT, and query timeouts.

• Concurrent client connections and server-side sessions.

• Support for native binaries, Docker containers, and Kubernetes.

• Integration with JDBC, ADBC, Python, Ibis, SQLAlchemy, and BI tools.

To be sure, GizmoData is a complete data platform with an elegant UI and pretty much offers what one would expect these days. Things like, Compute, Data, Notebooks, etc. But before we get all set on our own feet, let’s start with the open-source version and explore at its core what GizmoSQL is, and then later return to the Enterprise Cloud to compare and contrast.

Playing with open-source GizmoSQL.

One thing is clear when starting to familiarize oneself with the GizmoData GitHub repo(s) is that there is more than just a single tool. From what I can tell, and this just based on what I can read between the lines, they have built, and are building, an data ecosystem around the core product GizmoSQL Server. This makes sense and points to the fact that data vendors are learning their lessons, realizing that the hearts and minds of the data community is often won in the wider arena of integrations and implementations of any technology.

Sure, we want fast and slick tools. But, what we also want is …

• simplicity

• ease of use

• control

For example, Gizmo publishes an Operator specifically for running on Kubernetes … making less work for those wanting to easily deploy and run the tech stack.

Heck, they even provide an open-source UI for you to use and run if you want! I like this approach: be open and honest about what you are building, and if folk find it enticing and helpful, they could later turn into customers.

Let’s get back to the main core of the product GizmoSQL. Since it’s open source, let’s just play around with it using some Divvy Bike Trip data. We can build a little docker project around it to simply dip our toes in the water and see what happens, what kind of feelings we get when playing with it. This is an often missed but simple fact when identifying and researching new data tools of any kind. You can learn alot about the feel and ease with which you can play with “a thing.”

Here is a simple docker-compose.yml for our initial playing.

services:
  gizmosql:
    image: gizmodata/gizmosql:latest
    container_name: gizmosql
    init: true
    tty: true
    ports:
      - "31337:31337"
    environment:
      GIZMOSQL_USERNAME: gizmosql_user
      GIZMOSQL_PASSWORD: gizmosql_password
      TLS_ENABLED: "1"
      PRINT_QUERIES: "1"
      DATABASE_FILENAME: data/divvy.duckdb
    volumes:
      - ./data:/opt/gizmosql/data

Now you might be asking … sure … we have some sort of SQL Server running and DuckDB running somewhere, but how do I actually write code … or do a thing? I mean how do we run our SQL … the lingua franca of data? If you’ve been reading my Substack you will know I’m a big fan of Arrow, and the ADBC drivers built on arrow. We can use the gizmo ADBC driver to send our requests and get data back and forth.

import time
from adbc_driver_gizmosql import dbapi as gizmosql

CSV_GLOB = "/opt/gizmosql/data/*.csv"

QUERIES = {
    "Rows and date range across all CSV files": """
        SELECT count(*)                     AS trips,
               count(DISTINCT filename)     AS csv_files,
               min(started_at)              AS first_trip,
               max(started_at)              AS last_trip
        FROM trips
    """,
    "Members vs casual riders": """
        SELECT member_casual,
               count(*)                                                AS trips,
               round(avg(date_diff('second', started_at, ended_at)) / 60, 1) AS avg_minutes
        FROM trips
        GROUP BY ALL
        ORDER BY trips DESC
    """,
    "Bike types": """
        SELECT rideable_type, count(*) AS trips
        FROM trips
        GROUP BY ALL
        ORDER BY trips DESC
    """,
    "Top 10 start stations": """
        SELECT start_station_name, count(*) AS trips
        FROM trips
        WHERE start_station_name IS NOT NULL
        GROUP BY ALL
        ORDER BY trips DESC
        LIMIT 10
    """,
    "Busiest hours of the day": """
        SELECT hour(started_at) AS hour_of_day, count(*) AS trips
        FROM trips
        GROUP BY ALL
        ORDER BY trips DESC
        LIMIT 5
    """,
}


def main() -> None:
    with gizmosql.connect(
        "gizmosql://localhost:31337",
        username="gizmosql_user",
        password="gizmosql_password",
        tls_skip_verify=True,
    ) as conn, conn.cursor() as cur:
        cur.execute("SELECT version() AS duckdb_version")
        print(f"Connected to GizmoSQL — DuckDB {cur.fetchone()[0]}\n")

        cur.execute(f"""
            CREATE OR REPLACE VIEW trips AS
            SELECT * FROM read_csv('{CSV_GLOB}', union_by_name = true, filename = true)
        """)

        for title, sql in QUERIES.items():
            start = time.perf_counter()
            cur.execute(sql)
            table = cur.fetch_arrow_table()  # results arrive as Arrow over Flight SQL
            elapsed = time.perf_counter() - start
            print(f"== {title}  ({elapsed:.2f}s)")
            print(table.to_pandas().to_string(index=False))
            print()

if __name__ == "__main__":
    main()

Here are the results from this test run. Of course it is fast, one would expect that when putting DuckDB and Arrow in the same sentence.

To me, it isn’t that code itself that is the real secret, or special sauce, yet in another sense … it is. The point is that anyone coming from any data background can look at the code and say “Ah ha, I see, this is easy to use.” There is nothing “funny” that needs to be done to start using GizmoSQL … other than just installing it. As long as you have a client to connect to, you can start to submit work to that client as needed.

The real magic is hidden, it’s what you don’t see.

Clearly what GizmoData has done is provide the underlying infrastructure to supply the power of DuckDB and Arrow Flight SQL in a hardened client-server architecture that can be run or shipped anywhere. They're extending the power of something that can run on one person's laptop to a server with the kind of features you'd need to run it as a proper Data Platform.

The fact that we can reach critical mass where it’s now cheaper and easier to have a self-hosted SQL endpoint for when you’d otherwise reach for Snowflake … amazing! What a time to be alive. But, of course I wouldn’t be me if I didn’t try to break something. I always need to break something to feel good, or at least attempt it.

I’m curious, since this is at it’s core a server, I wonder what would happen if we tried to hammer it with writes. See what happens!

import argparse
import multiprocessing as mp
import os
import tempfile
import time
from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

import duckdb
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.csv as pacsv
from adbc_driver_gizmosql import dbapi as gizmosql

DATA_DIR = Path(__file__).parent / "data"
TARGET_TABLE = "trips_ingested"
LOG_TABLE = "ingest_log"

CONN_ARGS = dict(
    uri="gizmosql://localhost:31337",
    username="gizmosql_user",
    password="gizmosql_password",
    tls_skip_verify=True,
)
def connect():
    return gizmosql.connect(**CONN_ARGS)

def embedded_writer(db_path: str, worker: int, barrier) -> str:
    barrier.wait()
    try:
        con = duckdb.connect(db_path)
        con.execute("CREATE TABLE IF NOT EXISTS t (worker INT, ts TIMESTAMP)")
        con.execute("INSERT INTO t VALUES (?, now())", [worker])
        time.sleep(1)  # hold the connection open, like a real long-lived client
        con.close()
        return "ok"
    except duckdb.Error as e:
        return f"FAILED: {str(e).splitlines()[0][:110]}"


def part_a(workers: int) -> None:
    print(f"\n=== Part A: {workers} processes writing to ONE local DuckDB file directly ===")
    with tempfile.TemporaryDirectory() as tmp:
        db_path = str(Path(tmp) / "embedded.duckdb")
        with mp.Manager() as manager:
            barrier = manager.Barrier(workers)
            with ProcessPoolExecutor(max_workers=workers) as pool:
                results = list(pool.map(embedded_writer, [db_path] * workers, range(workers), [barrier] * workers))
    for i, r in enumerate(results):
        print(f"  process {i}: {r}")
    ok = results.count("ok")
    print(f"  -> {ok}/{workers} succeeded. DuckDB allows only one read-write process per file.")

_conn = None

def init_worker() -> None:
    global _conn
    _conn = connect()


def to_ipc(table: pa.Table) -> bytes:
    sink = pa.BufferOutputStream()
    with pa.ipc.new_stream(sink, table.schema) as w:
        w.write_table(table)
    return sink.getvalue().to_pybytes()


def ingest_batch(batch_id: int, payload: bytes) -> tuple[int, int, int, float]:
    batch = pa.ipc.open_stream(payload).read_all()
    start = time.perf_counter()
    with _conn.cursor() as cur:
        rows = cur.adbc_ingest(TARGET_TABLE, batch, mode="append")
        cur.execute(
            f"INSERT INTO {LOG_TABLE} VALUES (?, ?, ?, now())",
            parameters=[batch_id, os.getpid(), rows],
        )
    return batch_id, os.getpid(), rows, time.perf_counter() - start


def reader(stop, out) -> None:
    """Poll row counts while writers are running: reads are never blocked."""
    seen = []
    with connect() as conn, conn.cursor() as cur:
        while not stop.is_set():
            cur.execute(f"SELECT count(*) FROM {TARGET_TABLE}")
            seen.append(cur.fetchone()[0])
            time.sleep(0.05)
    out.put(seen)


def load_source(limit_files: int | None) -> pa.Table:
    files = sorted(DATA_DIR.glob("*.csv"))[:limit_files]
    if not files:
        raise SystemExit(f"No CSV files in {DATA_DIR}")
    tables = []
    for f in files:
        names = pacsv.open_csv(f).schema.names
        opts = pacsv.ConvertOptions(column_types={n: pa.string() for n in names}, strings_can_be_null=True)
        tables.append(pacsv.read_csv(f, convert_options=opts))
    print(f"Loaded {sum(t.num_rows for t in tables):,} rows from {len(files)} CSV file(s) on the client")
    return pa.concat_tables(tables, promote_options="default").combine_chunks()


def setup_tables(source: pa.Table) -> None:
    with connect() as conn, conn.cursor() as cur:
        cols = ", ".join(f'"{name}" VARCHAR' for name in source.schema.names)  # CSVs read as strings
        cur.execute(f"CREATE OR REPLACE TABLE {TARGET_TABLE} ({cols})")
        cur.execute(f"DROP TABLE IF EXISTS {LOG_TABLE}")
        cur.execute(f"CREATE TABLE {LOG_TABLE} (batch_id INT, pid INT, rows BIGINT, written_at TIMESTAMP)")


def part_b(source: pa.Table, workers: int, batch_rows: int, readers: int) -> tuple[list, float]:
    batches = [to_ipc(source.slice(off, batch_rows)) for off in range(0, source.num_rows, batch_rows)]
    print(f"\n=== Part B: {workers} processes x own ADBC session -> GizmoSQL ===")
    print(f"  {len(batches)} batches of up to {batch_rows:,} rows, {readers} reader process(es) polling")

    setup_tables(source)

    stop, out = mp.Event(), mp.Queue()
    reader_procs = [mp.Process(target=reader, args=(stop, out)) for _ in range(readers)]
    for p in reader_procs:
        p.start()

    results, errors = [], []
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=workers, initializer=init_worker) as pool:
        futures = [pool.submit(ingest_batch, i, b) for i, b in enumerate(batches)]
        for fut in as_completed(futures):
            try:
                results.append(fut.result())
            except Exception as e:  # report, don't hide
                errors.append(repr(e))
    elapsed = time.perf_counter() - start

    stop.set()
    reader_counts = [out.get() for _ in reader_procs]
    for p in reader_procs:
        p.join()

    print(f"  wrote {sum(r[2] for r in results):,} rows in {elapsed:.2f}s "
          f"({sum(r[2] for r in results) / elapsed:,.0f} rows/s, {len(results) * 2} write requests)")
    print(f"  errors: {len(errors)}")
    for e in errors[:5]:
        print(f"    {e}")
    lat = sorted(r[3] for r in results)
    print(f"  per-batch latency: p50 {lat[len(lat) // 2]:.3f}s, max {lat[-1]:.3f}s")
    for i, counts in enumerate(reader_counts):
        monotonic = all(a <= b for a, b in zip(counts, counts[1:]))
        print(f"  reader {i}: {len(counts)} reads during writes, "
              f"saw {counts[0]:,} -> {counts[-1]:,} rows, never went backwards: {monotonic}")
    return results, errors

def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--workers", type=int, default=8, help="writer processes")
    ap.add_argument("--batch-rows", type=int, default=5_000, help="rows per ingest request")
    ap.add_argument("--readers", type=int, default=2, help="reader processes polling during writes")
    ap.add_argument("--files", type=int, default=None, help="limit number of CSV files loaded")
    ap.add_argument("--skip-embedded", action="iframe.php?url=https%3A%2F%2Fdataengineeringcentral.substack.com%2Fstore_true", help="skip Part A")
    args = ap.parse_args()

    if not args.skip_embedded:
        part_a(args.workers)

    source = load_source(args.files)
    results, errors = part_b(source, args.workers, args.batch_rows, args.readers)
    expected = -(-source.num_rows // args.batch_rows)

Ok, so I’m a little disappointed I couldn’t break it, but again, DuckDB + Arrow Flight … what can I say.

GizmoData in the Cloud.

I wanted to also take a moment to look at managed GizmoData in the cloud incase you are scared of Kubernetes (you should be), or don’t know how to clone a git repo. The UI in the Cloud is easy to use, easy to sign up and get an account, and also comes with features I’m familar with after playing around with other data platforms.

Of course we have options to create Clusters, but the most interesting to me were the Data and Catalog sections. Clusters are clusters, but data integration is many times key to a platform. You would be surprised how many data teams just start with some data source(s) they want to use, and then wrap a data platform around that.

• We can see here that we can use and register DuckDB database files as first-class citizens.

• Also, we can create Catalogs for …

  • DuckLake and PostgreSQL

One of the biggest pieces I would like to see added would be first-class support for the two major Lakehouse data storage layers, Delta Lake and Apache Iceberg, especially via Unity Catalog and Iceberg catalogs like Polaris. DuckDB itself integrates perfectly with both Delta Lake and Iceberg, including catalogs like Unity; so adding them as options seems like it would make perfect sense.

That of course would not stop one from simply just integrating these data storage layers into your code, since DuckDB does support them.

Just as a sanity check, in our local GizmoSQL instance, I made sure we could talk to both Iceberg and Delta Lake, which in both cases worked fine.

All this thinking about wanting first-class support for catalogs like Unity as far as integration (because integration with a wide range of tech is always one of the most important things), I started to wonder about data pipelines. When I think of a data platform, I think about pipelines, many of them running at the same time, here and there … different pieces here and there tied together and running as a … well … DAG. Apache Airflow is the GOAT of data orchestrators … how could we run a task against a GizmoSQL Server?

from airflow.decorators import dag, task
from airflow.hooks.base import BaseHook
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1), catchup=False)
def divvy_daily():
      @task
      def run_sql(sql: str):
          from adbc_driver_gizmosql import dbapi as gizmosql
          c = BaseHook.get_connection("gizmosql_default")  # creds live in Airflow, not code
          with gizmosql.connect(f"gizmosql://{c.host}:{c.port}",
                                username=c.login, password=c.password) as conn, conn.cursor() as cur:
              cur.execute(sql)
      raw = run_sql.override(task_id="load_raw")(
          "CREATE OR REPLACE TABLE raw_trips AS SELECT * FROM read_parquet('s3://bucket/divvy/*.parquet')")
      agg = run_sql.override(task_id="station_daily")(
          "CREATE OR REPLACE TABLE station_daily AS SELECT start_station_name, started_at::date d, count(*) n FROM raw_trips GROUP BY ALL")
      raw >> agg
  divvy_daily()

Of course we could use the adbc driver for Gizmo and wire it into an Airflow DAG easy enough.

Of course, we could always switch that old friend … dbt, and wire up Gizmo that way.

uv add dbt-core dbt-gizmosql

I tested it out to make sure it would work. The profiles.yml points dbt at our GizmoSQL container over Arrow Flight SQL. Nothing exotic here, just a host, a port, and some credentials.

divvy:
  target: dev
  outputs:
    dev:
      type: gizmosql
      host: "{{ env_var('GIZMOSQL_HOST', 'localhost') }}"
      port: 31337
      database: divvy
      user: gizmosql_user
      password: "{{ env_var('GIZMOSQL_PASSWORD') }}"
      use_encryption: True
      tls_skip_verify: True 
      threads: 4

Then we write plain old dbt models. First, a staging view over the raw Divvy trips (the view sitting on top of our CSV files) …

- models/staging/stg_trips.sql
select
    ride_id,
    rideable_type,
    member_casual,
    start_station_name,
    end_station_name,
    started_at,
    ended_at,
    date_diff('second', started_at, ended_at) / 60.0 as ride_minutes
from {{ source('divvy_raw', 'trips') }}
where ended_at > started_at

… of course we could add models to our hearts desire, and then wire up dbt with Airflow to get the pipeline(s) running.

import os
import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

DBT_DIR = os.environ.get("DIVVY_DBT_DIR", "/opt/airflow/divvy_dbt")
DBT = f"cd {DBT_DIR} && dbt --no-use-colors"

with DAG(
    dag_id="divvy_dbt",
    schedule="@daily",
    start_date=pendulum.datetime(2026, 10, 1, tz="UTC"),
    catchup=False,
    tags=["gizmosql", "dbt"],
) as dag:
    dbt_run = BashOperator(task_id="dbt_run", bash_command=f"{DBT} run --profiles-dir .")
    dbt_test = BashOperator(task_id="dbt_test", bash_command=f"{DBT} test --profiles-dir .")

    dbt_run >> dbt_testimport os

import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

DBT_DIR = os.environ.get("DIVVY_DBT_DIR", "/opt/airflow/divvy_dbt")
DBT = f"cd {DBT_DIR} && dbt --no-use-colors"

with DAG(
    dag_id="divvy_dbt",
    schedule="@daily",
    start_date=pendulum.datetime(2026, 10, 1, tz="UTC"),
    catchup=False,
    tags=["gizmosql", "dbt"],
) as dag:
    dbt_run = BashOperator(task_id="dbt_run", bash_command=f"{DBT} run --profiles-dir .")
    dbt_test = BashOperator(task_id="dbt_test", bash_command=f"{DBT} test --profiles-dir .")

    dbt_run >> dbt_test

easy enough …

The rest of the GizmoData Enterprise account has more features that would appeal to a larger data team.

• Environments

• Secrets

• Extensions

• SSO / OAuth

• Catalog Permissions

• etc, etc.

Real Production workload on GizmoData. (remote s3 Iceberg Table)

If you want to read a pretty sweet benchmark, check out this 1 TB TPC-H Horse Race on GizmoData Cloud.

I really always try to do two types of tests when working with new tools: first, like we’ve already seen … do some local data testing to prove out the concepts and just see how things work. Secondly, it’s a good idea to just throw in some more realistic workloads, usually something that is simple on the surface, but forces a tool to do things like reading data over a network, etc. What I figured we could do in the GizmoData Cloud would be to run a large analytical query, maybe involving a window function on top of an Apache Iceberg table stored in S3.

Theoretically, this should be no problem for DuckDB. But what we are testing here would be the GizmoDB Cloud with a cluster we set up, and just see how it handles that real-world load of a large dataset across a network.

The truth is that in most production data teams, we have some sort of Lakehouse architecture backed by a remote storage system like S3, many times involving something like Hudi, Delta, or Iceberg. We would be remiss not to test this out.

• For this test, we will use the open-source Backblaze Harddrive Dataset. They provide it as an Iceberg table on s3.

First, I had to set up a Cluster; it’s pretty standard stuff, just selecting what you want.

Now that my cluster is running. I can get the endpoint + the user and password I created when building that cluster, and hopefully run the script below against it.

Anyways, let’s write a Python ABDC driver program that can run an analytical query over a bunch of this data, and then persist the results to a DuckDB table. Ok, this is going to be a mouthful, but again, we are wanting to do something real here, so bear with me.

import argparse
import os
import time
from datetime import date
from pathlib import Path

import duckdb
import pyarrow as pa
from adbc_driver_gizmosql import dbapi as gizmosql

# Read-only key published by Backblaze for the public Drive Stats bucket
B2_KEY_ID = os.environ.get("B2_KEY_ID", "0045f0571db506a0000000017")
B2_SECRET = os.environ.get("B2_SECRET", "K004Fs/bgmTk5dgo6GAVm2Waj3Ka+TE")
B2_REGION = "us-west-004"
B2_ENDPOINT = "s3.us-west-004.backblazeb2.com"
TABLE_ROOT = "s3://drivestats-iceberg/drivestats"

LOCAL_DB = Path(__file__).parent / "data" / "drivestats_results.duckdb"
RESULT_TABLE = "model_monthly_health"

CONN_ARGS = dict(
    uri=os.environ.get("GIZMOSQL_URI", "gizmosql://benchmark-97f775ae655c49188bd5159d953c4106.us-east-2.aws.sql.gizmodata.com:443"),
    username=os.environ.get("GIZMOSQL_USERNAME", "gizmosql_admin"),
    password=os.environ.get("GIZMOSQL_PASSWORD", "bingbongdingdong"),
    tls_skip_verify=os.environ.get("GIZMOSQL_TLS_SKIP_VERIFY", "0") == "1",  # set to 1 for the local container's self-signed cert
)

# Per drive, per day: window functions over each drive's daily history, then
# a rollup to model x month. RANGE frames use calendar days, so gaps in a
# drive's reporting don't stretch the windows.
#
# The query runs one chunk of whole months at a time so it fits in server
# memory. Every frame reaches at most 30 days back or forward, so each chunk
# scans MARGIN_DAYS of extra data on both sides, then keeps only its own
# months. Chunks never share a month, so their outputs just concatenate.
MARGIN_DAYS = 31
WINDOW_SQL = """
WITH daily AS (
    SELECT date, serial_number, model, capacity_bytes, failure,
           smart_5_raw   AS reallocated_sectors,
           smart_9_raw   AS power_on_hours,
           smart_194_raw AS temperature_c
    FROM drivestats
    WHERE date >= DATE '{start}' - INTERVAL {margin} DAY AND date < DATE '{end}' + INTERVAL {margin} DAY
),
per_drive AS (
    SELECT *,
           reallocated_sectors - lag(reallocated_sectors) OVER d                                      AS realloc_delta,
           avg(temperature_c)          OVER (d RANGE BETWEEN INTERVAL 6 DAYS PRECEDING AND CURRENT ROW) AS temp_7d_avg,
           max(temperature_c)          OVER (d RANGE BETWEEN INTERVAL 29 DAYS PRECEDING AND CURRENT ROW) AS temp_30d_max,
           max(failure)                OVER (d RANGE BETWEEN CURRENT ROW AND INTERVAL 29 DAYS FOLLOWING) AS fails_within_30d
    FROM daily
    WINDOW d AS (PARTITION BY serial_number ORDER BY date)
)
SELECT model,
       date_trunc('month', date)::DATE                                        AS month,
       any_value(capacity_bytes) / 1e12                                       AS capacity_tb,
       count(*)                                                               AS drive_days,
       count(DISTINCT serial_number)                                          AS drives,
       sum(failure)                                                           AS failures,
       round(sum(failure) * 365.0 * 100 / count(*), 3)                        AS annualized_failure_rate_pct,
       round(avg(temp_7d_avg), 2)                                             AS avg_temp_7d,
       max(temp_30d_max)                                                      AS max_temp_30d,
       round(avg(power_on_hours) / 24 / 365, 2)                               AS avg_age_years,
       count(*) FILTER (WHERE realloc_delta > 0)                              AS realloc_growth_days,
       count(*) FILTER (WHERE fails_within_30d = 1)                           AS pre_failure_drive_days,
       round(avg(reallocated_sectors) FILTER (WHERE fails_within_30d = 1), 1) AS avg_realloc_pre_failure,
       round(avg(reallocated_sectors) FILTER (WHERE fails_within_30d = 0), 1) AS avg_realloc_healthy
FROM per_drive
WHERE date >= DATE '{start}' AND date < DATE '{end}'
GROUP BY ALL
ORDER BY month, model
"""


def setup_source(cur) -> str:
    """Point the server session at the Iceberg table and return the metadata file used."""
    cur.execute("INSTALL httpfs; LOAD httpfs; INSTALL iceberg; LOAD iceberg;")
    cur.execute(f"""
        CREATE OR REPLACE SECRET drivestats_b2 (
            TYPE s3, KEY_ID '{B2_KEY_ID}', SECRET '{B2_SECRET}',
            REGION '{B2_REGION}', ENDPOINT '{B2_ENDPOINT}', URL_STYLE 'path',
            SCOPE '{TABLE_ROOT}'
        )
    """)
    # No version-hint.text in the bucket: pick the newest metadata file.
    # Names start with a zero-padded sequence number, so max() is the latest.
    cur.execute(f"SELECT max(file) FROM glob('{TABLE_ROOT}/metadata/*.metadata.json')")
    metadata = cur.fetchone()[0]
    cur.execute(f"CREATE OR REPLACE TEMP VIEW drivestats AS SELECT * FROM iceberg_scan('{metadata}')")
    return metadata


def show(cur, title: str, sql: str, max_rows: int = 50) -> None:
    start = time.perf_counter()
    cur.execute(sql)
    df = cur.fetch_arrow_table().to_pandas()
    print(f"\n== {title}  ({time.perf_counter() - start:.2f}s)")
    print(df.to_string(index=False, max_rows=max_rows))


def stats(cur, metadata: str, start: str, end: str) -> None:
    # File-level row counts come from the manifests, so nothing is scanned here.
    # Partition values are month strings, so filter on the date_month= path part.
    show(cur, f"Rows to process, {start} to {end} (from Iceberg manifests)", f"""
        WITH files AS (
            SELECT regexp_extract(file_path, 'date_month=([0-9-]+)', 1) AS month, record_count
            FROM iceberg_metadata('{metadata}')
            WHERE status <> 'DELETED'
        )
        SELECT left(month, 4) AS year, count(*) AS data_files, sum(record_count) AS rows
        FROM files
        WHERE month >= left('{start}', 7) AND month < left('{end}', 7)
        GROUP BY ALL
        UNION ALL
        SELECT 'TOTAL', count(*), sum(record_count)
        FROM files
        WHERE month >= left('{start}', 7) AND month < left('{end}', 7)
        ORDER BY year
    """)
    show(cur, "Whole table (current snapshot)", f"""
        SELECT count(*) AS data_files, sum(record_count) AS rows,
               min(regexp_extract(file_path, 'date_month=([0-9-]+)', 1)) AS first_month,
               max(regexp_extract(file_path, 'date_month=([0-9-]+)', 1)) AS last_month
        FROM iceberg_metadata('{metadata}')
        WHERE status <> 'DELETED'
    """)

    cur.execute("DESCRIBE drivestats")
    schema = cur.fetch_arrow_table().to_pandas()
    smart = schema[schema.column_name.str.startswith("smart_")]
    other = schema[~schema.column_name.str.startswith("smart_")]
    print(f"\n== Schema: {len(schema)} columns ({len(smart)} SMART attributes)")
    print(other[["column_name", "column_type"]].to_string(index=False))
    print(f"  + {len(smart)} smart_<id>_raw / smart_<id>_normalized BIGINT columns")

    show(cur, "Sample rows (columns used by the benchmark)", f"""
        SELECT date, serial_number, model, capacity_bytes, failure,
               smart_5_raw, smart_9_raw, smart_194_raw, datacenter
        FROM drivestats
        WHERE date = DATE '{start}'
        LIMIT 10
    """)


def month_chunks(start: str, end: str, months: int) -> list[tuple[str, str]]:
    """Split [start, end) into consecutive ranges of whole months."""
    chunks, cur, stop = [], date.fromisoformat(start), date.fromisoformat(end)
    while cur < stop:
        m = cur.month - 1 + months
        nxt = min(date(cur.year + m // 12, m % 12 + 1, 1), stop)
        chunks.append((cur.isoformat(), nxt.isoformat()))
        cur = nxt
    return chunks


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--start", default="2023-01-01", help="inclusive, YYYY-MM-DD (default %(default)s)")
    parser.add_argument("--end", default="2026-01-01", help="exclusive, YYYY-MM-DD (default %(default)s)")
    parser.add_argument("--skip-stats", action="iframe.php?url=https%3A%2F%2Fdataengineeringcentral.substack.com%2Fstore_true", help="only run the benchmark query")
    parser.add_argument("--chunk-months", type=int, default=3,
                        help="months per server query; lower it if the server runs out of memory (default %(default)s)")
    args = parser.parse_args()

    total_start = time.perf_counter()
    with gizmosql.connect(**CONN_ARGS) as conn, conn.cursor() as cur:
        cur.execute("SELECT version()")
        print(f"Connected to {CONN_ARGS['uri']} (DuckDB {cur.fetchone()[0]})")
        cur.execute("SELECT current_setting('threads'), current_setting('memory_limit')")
        threads, mem = cur.fetchone()
        print(f"Server: {threads} threads, memory_limit {mem}")

        metadata = setup_source(cur)
        print(f"Iceberg metadata: {metadata}")

        if not args.skip_stats:
            stats(cur, metadata, args.start, args.end)

        cur.execute("SET preserve_insertion_order = false")  # lets DuckDB spill more freely
        chunks = month_chunks(args.start, args.end, args.chunk_months)
        print(f"\n== Benchmark: window functions per drive, {args.start} to {args.end}"
              f" ({len(chunks)} chunks of {args.chunk_months} months, {MARGIN_DAYS}-day margins)")
        query_start = time.perf_counter()
        parts = []
        for chunk_start, chunk_end in chunks:
            t = time.perf_counter()
            cur.execute(WINDOW_SQL.format(start=chunk_start, end=chunk_end, margin=MARGIN_DAYS))
            parts.append(cur.fetch_arrow_table())  # Arrow over Flight SQL
            print(f"  {chunk_start} to {chunk_end}: {time.perf_counter() - t:7.2f}s"
                  f"  {sum(parts[-1].column('drive_days').to_pylist()):>12,} drive-days")
        result = pa.concat_tables(parts)
        query_secs = time.perf_counter() - query_start

    write_start = time.perf_counter()
    LOCAL_DB.parent.mkdir(exist_ok=True)
    with duckdb.connect(str(LOCAL_DB)) as local:
        local.execute(f"CREATE OR REPLACE TABLE {RESULT_TABLE} AS SELECT * FROM result")
        rows_in, drive_days, failures = local.execute(
            f"SELECT count(*), sum(drive_days)::BIGINT, sum(failures)::BIGINT FROM {RESULT_TABLE}"
        ).fetchone()
        top = local.execute(f"""
            SELECT model, sum(drive_days)::BIGINT AS drive_days, sum(failures)::BIGINT AS failures,
                   round(sum(failures) * 365.0 * 100 / sum(drive_days), 2) AS afr_pct
            FROM {RESULT_TABLE}
            GROUP BY model
            HAVING sum(drive_days) > 1000000
            ORDER BY afr_pct DESC
            LIMIT 10
        """).df()
    write_secs = time.perf_counter() - write_start

    print(f"  server scan + windows + fetch: {query_secs:.2f}s")
    print(f"  local DuckDB write:            {write_secs:.2f}s")
    print(f"  total wall clock:              {time.perf_counter() - total_start:.2f}s")
    print(f"  source rows processed:         {drive_days:,} drive-days, {failures:,} failures")
    print(f"  result:                        {rows_in:,} rows -> {LOCAL_DB}:{RESULT_TABLE}")
    print("\n== Highest annualized failure rate (models with >1M drive-days in range)")
    print(top.to_string(index=False))


if __name__ == "__main__":
    main()

So, GizmoData Cloud gives you a ton of information and metrics about a running Cluster, how to connect to it from a variety of sources, etc. I was also able to check the Pod Logs in the UI and see a bunch of info; you can see our query running in the logs there. So we know it’s working!

I can also watch memory consumption on the cluster.

So now, we just have to wait and get the results back.

 Monitor event: 2023-01-01 to 2023-04-01: 63.64s 21,454,992 drive-days · cloud benchmark chunk progress
⏺ The first chunk finished: Q1 2023, 21.5M drive-days in 63.6s.
⏺ Monitor event: 2023-04-01 to 2023-07-01: 55.65s 21,835,156 drive-days · cloud benchmark chunk progress
⏺ Chunk 2 of 12 (Q2 2023) finished: 21.8M drive-days in 55.7s.
⏺ Monitor event: 2023-07-01 to 2023-10-01: 60.30s 23,601,986 drive-days · cloud benchmark chunk progress
⏺ Chunk 3 of 12 (Q3 2023) took 60.3s and covered 23.6M drive-days.
⏺ Monitor event: 2023-10-01 to 2024-01-01: 58.00s 24,790,999 drive-days · cloud benchmark chunk progress
⏺ Chunk 4 of 12 (Q4 2023) finished in 58.0s, covering 24.8M drive-days.
⏺ Monitor event: 2024-01-01 to 2024-04-01: 61.19s 25,189,213 drive-days · cloud benchmark chunk progress
⏺ Chunk 5 of 12 (Q1 2024) finished in 61.2s and processed 25.2M drive-days. 
⏺ Monitor event: 2024-04-01 to 2024-07-01: 64.88s 26,022,269 drive-days · cloud benchmark chunk progress
⏺ Chunk 6 of 12 (Q2 2024) finished in 64.9s, 26.0M drive-days.
⏺ Monitor event: 2024-07-01 to 2024-10-01: 65.45s 26,822,987 drive-days · cloud benchmark chunk progress
⏺ Chunk 7 of 12 (Q3 2024) finished in 65.5s and covered 26.8M drive-days.
⏺ Monitor event: 2024-10-01 to 2025-01-01: 87.33s 27,345,292 drive-days · cloud benchmark chunk progress
⏺ Chunk 8 of 12 (Q4 2024) finished: 27.3M drive-days in 87.3s.
⏺ Monitor event: 2025-01-01 to 2025-04-01: 128.81s 27,799,986 drive-days · cloud benchmark chunk progress
⏺ Chunk 9 of 12 (Q1 2025) finished: 27.8M drive-days in 128.8s.
⏺ Monitor event: 2025-04-01 to 2025-07-01: 137.52s 28,808,042 drive-days · cloud benchmark chunk progress
⏺ Chunk 10 of 12 (Q2 2025) finished: 137.5s, 28.8M drive-days.
⏺ Monitor event: 2025-07-01 to 2025-10-01: 141.31s 29,844,451 drive-days · cloud benchmark chunk progress
⏺ Chunk 11 of 12 (Q3 2025) finished in 141.3s, covering 29.8M drive-days. One chunk left.
⏺ Monitor event: 2025-10-01 to 2026-01-01: 137.82s 30,941,708 drive-days

• The 3-year run took 17 min 54 s (1,073.5s) on the 2-thread, 7.4 GiB Core cluster.

Well, if you look at the script, we made that small cluster work for its food! This is one of the things I love about these new DuckDB- and Arrow-based tools: you can run real workloads … hard ones … on affordable commodity hardware. They can handle it!

This is where you can really tell what a tool is made of. Lots of people will publish and release things with simple “Hello World” type benchmarks or local examples. The truth is you need to stress test a system.

• Throw cloud storage at it

• Use small clusters to work on oversized data to see if it pukes

• Use Iceberg or some other storage layer

The time to test this sort of real-world use case isn’t after the fact.

Thoughts on GizmoData

I was pretty excited to try out Gizmo, especially given its large open-source footprint on GitHub. This is always one of the first things I look for, not that open-source is special … but it is. It lets us try something out and use it as much as we want, get to know it, understand it … in a way many companies don’t allow.

Take DuckDB, wrap it in Arrow Flight SQL, harden it into a real client-server setup, and hand it to me. What more could a data guy ask for!? Within a few minutes of docker compose up, I was running SQL over Divvy CSV files, querying from Python through ADBC, and clicking around in an open-source UI. Then I did what I always do and tried to break it.

Eight processes hammering it with writes and readers polling the whole time … and nothing broke. Add in dbt models, an Airflow DAG, and a quick sanity check that Iceberg and Delta Lake play nicely through DuckDB, and you start to see the picture.

I’m looking forward to seeing this platform grow and morph into something even bigger in the future. Some of the things I will be watching for in the future …

• First-class support for Unity and Polaris-type catalogs

  • and then, by extension, Iceberg and Delta Lake

• First-class support for data pipelines

  • could be from Airflow Operators or its own scheduling and DAG tool

As the days and weeks trickle by and fall to the ground like the autumn leaves on the maple tree behind the house, my mind wanders back to the good ole’ days when I had nothing in front of me except Notepad++, a terminal, and a problem to solve by myself. Those days are long gone, and Sir Claudius is threatening to turn me into a milquetoast programmer. Some days I feel like screaming (in all caps) at Claudius… “Don’t you know I used to hand-roll Perl before you were born??!!!” Alas, I must find other ways to rage against the machine.

To keep my fingers young and flexible, and the grey matter inside my noggin pliable … I’ve written a few posts about going back to the basics here and here, as well as how to keep our skills sharp in the age of AI. Today builds on that topic. Not only are we going to write Rust by hand, but we are also going to flex the creativity muscles that seem to atrophy as quickly as they release new foundational models.

We will talk about something that might be new to most of you, the Delta Kernel, and then play with it in Rust, doing something useful.

What and why of Delta Kernel.

For anyone who’s played around in the Lakehouse world at all, you will be very familiar with Delta Lake, one of the premier ways to build on data, a de facto standard for ACID interaction at scale. Of course, this rise in popularity of the Lakehouse architecture, combined with tools like Delta Lake to sit on top of storage systems like S3, has not been without its bumps in the road. These … “opportunities for improvement” have nothing to do with Delta itself, but with its popularity and the broader data tooling and community.

What happens when you have a large ecosystem of tooling, all running on their own clocks, using a single data layer?

Well, of course you run into many problems as the protocol changes, upgrades, improves, adds features, etc. I’ve had these issues for years when using tools like DuckDB and Polars in production, interacting with Databricks Delta Lake tables managed in and out of Unity. Versions change … deletion vectors are added; tools stop working … all of this is understandable. It’s hard for different orgs, probably overworked open-source hobbits, to keep up to date with each other.

• This is what Delta Kernel aims to solve.

There are two versions of the Delta Kernel, and they are made to match the wider data ecosystems and the languages behind the tooling of the future; Rust and Java. Part of the challenge external tools face when working with a Lakehouse and a protocol like the Delta Protocol is handling metadata, raw data, checkpoints, and all the necessary complexity that makes this technology possible. So, Delta Kernel gives tool builders and integrators a one-stop shop option to avoid some of that complexity.

Of course, we are going to test out that theory today by building a tool in Rust.

Trying out Rust + Delta Kernel.

Ok, so the best way to learn the power of what the Delta Kernel can do for us is to thumb our nose at Claudius Maximus and build a tool ourselves to see what we can learn. Before we do that, let’s figure out the basic concepts behind the APIs Delta Kernel offers. Hopefully, we will find that it makes common Lakehouse functions very easy to perform. We will also need to come up with some little tool to build, something useful that will give the reason to write some Rust.

Also, remember that the Delta Kernel is probably not for everyday use, but rather for those who want to build tools and integrations into existing tools.

I must say that the GitHub for the Delta Kernel Rust says it’s experimental, especially write, which is disappointing, but hopefully all comes to fruition.

The blog post that I was reading for my introduction to the Delta Kernel was already out of date. Sounds like things started with an easier Table API to read/write data, but that’s no more. Things are changing fast, and that’s ok; we will get through by hook or crook.

Let’s see if we can simply write some Rust code to read an existing Delta Lake table on S3. As far as I understand, what will happen when using the Delta Kernel for Rust … is this …

• Engine is how to deal with Delta

• Snapshot , Scan, or Transaction APIs for read/write

Let’s say that we work for some consultant company that does a lot of Databricks and/or Delta Lake migrations of one sort or another. Play along with me here. We are some hobbits living in our mother's basement and are getting tired of trying to pull data in and out of Delta Lake to go between different accounts and systems. We need an Import and Export tool.

cargo init delta-import-export

Let’s start by dipping our toes in the water and simply reading a Delta Lake table stored in S3, see what the code looks like.

cargo add delta_kernel
cargo add delta_kernel_default_engine --features rustls
cargo add clap --features derive
cargo add arrow
cargo add url
cargo add parquet --features arrow

Ok, we have some of our core libraries we will use; next, let’s use clap as a way to build our import/export tool with some CLI command-line args that make sense in our case. Like s3 location of a Delta table, parquet file location for data in/out, etc.

This will be our lib.rs

use clap::{Parser, Subcommand};
use std::path::PathBuf;

#[derive(Parser, Debug)]
#[command(
    name = "delta-import-export",
    version,
    about = "Import and export Delta Lake tables"
)]
pub struct Args {
    #[command(subcommand)]
    pub command: Command,
}

#[derive(Subcommand, Debug)]
pub enum Command {
    Export {
        #[arg(short, long)]
        table: String,

        #[arg(short, long)]
        output: PathBuf,
    },

    Import {
        #[arg(short, long)]
        table: String,

        #[arg(short, long)]
        parquet: PathBuf,
    },
}

And our main.rs stubbed out; we will talk about this all later. Honestly, it’s not that much code at all, and pretty much captures what I expected. The reason there is more code than you might think is because I’m not just printing the results of the Delta Kernel code to stdout; I’m actually using Arrow and Parquet crates, plus some logic with clap to capture command line args, and also push the results of the Delta table with ArrowBatch records out to a parquet file.

use std::fs::File;
use std::path::PathBuf;
use std::sync::Arc;

use clap::Parser;
use delta_kernel::engine::arrow_data::EngineDataArrowExt;
use delta_kernel::object_store::aws::AmazonS3Builder;
use delta_kernel::Snapshot;
use delta_kernel_default_engine::DefaultEngineBuilder;
use parquet::arrow::ArrowWriter;
use parquet::basic::Compression;
use parquet::file::properties::WriterProperties;

#[derive(Parser, Debug)]
#[command(
    name = "delta-export",
    version,
    about = "Export an S3 Delta table to a local Parquet file"
)]
struct Args {
    #[arg(short, long)]
    table: String,

    #[arg(short, long)]
    output: PathBuf,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::parse();

    println!("Delta table: {}", args.table);
    println!("Output file: {}", args.output.display());

    let table_url = delta_kernel::try_parse_uri(&args.table)?;
    let object_store = AmazonS3Builder::from_env()
        .with_url(&args.table)
        .build()?;
    let engine = DefaultEngineBuilder::new(Arc::new(object_store)).build();

    let snapshot = Snapshot::builder_for(&table_url).build(&engine)?;
    println!("Delta version: {}", snapshot.version());
    let scan = snapshot.scan_builder().build()?;
    let batches = scan.execute(Arc::new(engine))?;

    let mut parquet_writer = None;
    let mut total_rows = 0usize;
    let mut batch_count = 0usize;

    for engine_data in batches {
        let batch = EngineDataArrowExt::try_into_record_batch(engine_data)?;

        if parquet_writer.is_none() {
            let file = File::create(&args.output)?;

            let properties = WriterProperties::builder()
                .set_compression(Compression::SNAPPY)
                .build();

            parquet_writer = Some(ArrowWriter::try_new(
                file,
                batch.schema(),
                Some(properties),
            )?);
        }

        if let Some(writer) = parquet_writer.as_mut() {
            writer.write(&batch)?;
        }

        total_rows += batch.num_rows();
        batch_count += 1;

        println!(
            "Batch {}: {} rows | total: {}",
            batch_count,
            batch.num_rows(),
            total_rows
        );
    }

    if let Some(writer) = parquet_writer {
        writer.close()?;
    }

    println!("Rows written: {}", total_rows);
    println!("Output: {}", args.output.display());

    Ok(())
}

Here is a Delta Lake table for hobbits.

This table is located at s3://confessions-of-a-data-guy/hobbits. Let’s get to the core of writing some simple code to read this Delta table we will pass in, and stream those records with Arrow out to a parquet file locally.

cargo run -- \
  --table s3://confessions-of-a-data-guy/hobbits \
  --output ./hobbits.parquet

...

danielbeach@Daniels-MacBook-Pro-2 delta-import-export % cargo run -- \
  --table s3://confessions-of-a-data-guy/hobbits \
  --output ./hobbits.parquet
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.35s
     Running `target/debug/delta-import-export --table 's3://confessions-of-a-data-guy/hobbits' --output ./hobbits.parquet`
Delta table: s3://confessions-of-a-data-guy/hobbits
Output file: ./hobbits.parquet
Delta version: 1
Batch 1: 8 rows | total: 8
Rows written: 8
Output: ./hobbits.parquet

There we go! We have less than 100 lines of Rust code all in, and we can export a remote S3 Delta table to a parquet file. Of course, in the real world we would want to control batch sizes, etc in massive tables, but you get the drift. Of course, a little later we can add some code to do the opposite, maybe push local files CSV, parquet, etc.) back up to a brand new or existing Delta table. Let’s talk about some of the Delta Kernel API code we see exposed to us.

• First, we can see how easy the Kernel made it to use the Engine to set up a remote S3 located bunch of tables without much fuss. This is a good thing because this is the real world after all. Nothing is worse than a good tool made to work locally that chokes on reality.

let table_url = delta_kernel::try_parse_uri(&args.table)?;

let object_store = AmazonS3Builder::from_env()
        .with_url(&args.table)
        .build()?;

let engine = DefaultEngineBuilder::new(Arc::new(object_store)).build()

Things that are happening here behind the hood are all the nitty-gritty details of dealing with the Delta Protocol itself. Remember, we are dealing with a layer on top of a storage engine with multiple files: data files, metadata files, checkpoints, etc., all stored in specific places and structures that need to be interacted with in a certain way. The Kernel is doing all that for us … on S3! This makes our life wayyyy easier.

• Where is the Delta Table located

• How do I talk to S3 to get what I need?

• Build an Engine for the Kernel that can do this talk.

Next, now that we have an Engine ready to go, we can do the very obvious task that is not surprising at all to anyone who’s played around with Arrow or any modern Lakehouse architecture.

let snapshot = Snapshot::builder_for(&table_url).build(&engine)?;
println!("Delta version: {}", snapshot.version());
let scan = snapshot.scan_builder().build()?;
let batches = scan.execute(Arc::new(engine))?;

We basically get the latest snapshot of the Delta table; in theory, we could get any point in time … scan it, so basically build a query, and then get an iterator on those results in batches. Pretty easy and obvious stuff. This is what makes the Kernel so powerful, as you can see. Think about how much easier this makes our life if we are maintainers of DuckDB , Polars, Daft or any other tool, like the one we are building. We don’t need to become experts in the Delta Protocol just to read and write some data. The Kernel does the hard work; we can just worry about integration into our tool.

I don’t really feel like going through the rest of the code to talk about how we flush those batches to parquet, that is just normal io and file work stuff.

Let’s add some functionality to create a new Delta table and do the reverse, flush local records up to a remote table. This should show us what it’s like to do a Transaction for a Lakehouse, which is an extremely important feature and topic. We all like READ tools, but WRITE is where the power is at.

use std::fs::File;
use std::path::PathBuf;
use std::sync::Arc;

use arrow_csv::reader::Format;
use arrow_csv::ReaderBuilder;
use clap::Parser;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::arrow_conversion::TryIntoKernel;
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::object_store::aws::AmazonS3Builder;
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::{CommitResult, RetryableTransaction};
use delta_kernel::Snapshot;
use delta_kernel_default_engine::DefaultEngineBuilder;

#[derive(Parser, Debug)]
#[command(name = "delta-import", version)]
struct Args {
    #[arg(short, long)]
    table: String,

    #[arg(short, long)]
    csv: PathBuf,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::parse();

    println!("CSV: {}", args.csv.display());
    println!("Delta table: {}", args.table);

    let table_url = delta_kernel::try_parse_uri(&args.table)?;
    let object_store = AmazonS3Builder::from_env()
        .with_url(&args.table)
        .build()?;

    let engine = DefaultEngineBuilder::new(Arc::new(object_store)).build();
    let schema_file = File::open(&args.csv)?;
    let format = Format::default().with_header(true);
    let (arrow_schema, _) = format.infer_schema(schema_file, Some(1000))?;

    println!("Schema: {arrow_schema:#?}");

    let kernel_schema = Arc::new(
        (&arrow_schema).try_into_kernel()?
    );

    println!("Creating Delta table...");

    create_table(
        &args.table,
        kernel_schema,
        "delta-import-export/0.1.0",
    )
    .build(
        &engine,
        Box::new(FileSystemCommitter::new()),
    )?
    .commit(&engine)?;

    println!("Delta table created.");

    let snapshot = Snapshot::builder_for(table_url.clone())
        .build(&engine)?;

    let mut transaction = snapshot
        .transaction(
            Box::new(FileSystemCommitter::new()),
            &engine,
        )?
        .with_operation("WRITE".to_string())
        .with_engine_info("delta-import-export/0.1.0")
        .with_data_change(true);

    let write_context = transaction
        .write_state()?
        .write_context_builder()
        .build()?;

    let csv_file = File::open(&args.csv)?;

    let csv_reader = ReaderBuilder::new(
        Arc::new(arrow_schema)
    )
    .with_header(true)
    .with_batch_size(8192)
    .build(csv_file)?;

    let mut total_rows = 0usize;

    for batch in csv_reader {
        let batch = batch?;
        total_rows += batch.num_rows();
        println!("Writing batch: {} rows", batch.num_rows());
        let data = ArrowEngineData::new(batch);

        let file_metadata = engine
            .write_parquet(
                &data,
                &write_context,
            )
            .await?;
        transaction.add_files(file_metadata);
    }

    let mut retries = 0;
    let committed = loop {
        if retries > 5 {
            return Err("Exceeded maximum commit retries".into());
        }
        transaction = match transaction.commit(&engine)? {
            CommitResult::Committed(committed) => {
                break committed;
            }
            CommitResult::Conflicted(conflicted) => {
                return Err(
                    format!(
                        "Transaction conflict at version {}",
                        conflicted.conflict_version()
                    )
                    .into()
                );
            }
            CommitResult::Retryable(
                RetryableTransaction {
                    transaction,
                    error,
                }
            ) => {
                println!("Retrying commit: {error}");
                transaction
            }
        };
        retries += 1;
    };

    println!(
        "Committed Delta version: {}",
        committed.commit_version()
    );
    println!("Rows written: {}", total_rows);
    Ok(())
}

And let’s run it before we talk about the code.

danielbeach@Daniels-MacBook-Pro-2 delta-import-export % export AWS_DEFAULT_REGION="us-east-1"
cargo run -- \
  --table s3://confessions-of-a-data-guy/hobbits-new \
  --csv ./hobbits.csv
  --> src/main.rs:56:5
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.35s
     Running `target/debug/delta-import-export --table 's3://confessions-of-a-data-guy/hobbits-new' --csv ./hobbits.csv`
CSV: ./hobbits.csv
Delta table: s3://confessions-of-a-data-guy/hobbits-new
Schema: Schema {
    fields: [
        Field {
            name: "hobbit_id",
            data_type: Int64,
            nullable: true,
        },
        Field {
            name: "name",
            data_type: Utf8,
            nullable: true,
        },
        Field {
            name: "family_name",
            data_type: Utf8,
            nullable: true,
        },
        Field {
            name: "age",
            data_type: Int64,
            nullable: true,
        },
        Field {
            name: "home",
            data_type: Utf8,
            nullable: true,
        },
        Field {
            name: "occupation",
            data_type: Utf8,
            nullable: true,
        },
        Field {
            name: "favorite_food",
            data_type: Utf8,
            nullable: true,
        },
        Field {
            name: "ring_bearer",
            data_type: Boolean,
            nullable: true,
        },
        Field {
            name: "adventures",
            data_type: Int64,
            nullable: true,
        },
        Field {
            name: "last_updated",
            data_type: Timestamp(
                Millisecond,
                None,
            ),
            nullable: true,
        },
    ],
    metadata: {},
}
Creating Delta table...
Delta table created.
Writing batch: 8 rows
Committed Delta version: 1
Rows written: 8

We should double-check S3 to see if it’s actually there, which indeed it is.

Let’s talk about this code a little bit; of course, we can expect it to be a little more complicated because we are, after all, making a commit into a Delta table. Let’s not talk about race conditions and pushing commits through Unity Catalog so we can deal with the reality of multiple writers. That’s for another day.

It’s very easy to create a table, which is nice.

create_table(
        &args.table,
        kernel_schema,
        "delta-import-export/0.1.0",
    )
    .build(
        &engine,
        Box::new(FileSystemCommitter::new()),
    )?
    .commit(&engine)?;

Next, we had to do the work of snapshot and transaction.

let snapshot = Snapshot::builder_for(table_url.clone())
        .build(&engine)?;

let mut transaction = snapshot
        .transaction(
            Box::new(FileSystemCommitter::new()),
            &engine,
        )?
        .with_operation("WRITE".to_string())
        .with_engine_info("delta-import-export/0.1.0")
        .with_data_change(true);

let write_context = transaction
        .write_state()?
        .write_context_builder()
        .build()?;

Pretty straightforward even though it might look like a mess; this isn’t really too bad to make a commit into the transaction log for Delta Lake. We just grab a snapshot and do the WRITE operation.

transaction = match transaction.commit(&engine)? {
            CommitResult::Committed(committed) => {
                break committed;
            }

Of course we gotta do retries and all that jazz in the production code. I think the bigger picture of what is going on here is more important than the details or the code. What we’ve seen here is the very basics; if you are an astute user of the Lakehouse, the first question might be … what about real Unity Catalog Lakehouses … what about multiple writers in large systems?

• Never fear, it appears there are some Unity Catalog crates to deal with just this problem.

So our code still does its job, but the Unity crates would add that extra layer of support. Maybe we will get to that later.

The real winner-winner chicken dinner.

Sure, we could keep walking down this road, adding more features to my little CLI tool, getting Unity crates wired up so we can hit production tables with read/writes and not worry about losing our job. But I think that the more exciting and less talked about topic is the fact that this even exists in the first place. Let me get up on my soapbox and preach to you for a moment.

Let me ask you a question, and I will answer it for you. When was the last time you saw such a production-ready, hardened, and open-source Lake House architecture as what we are discussing here Delta Kernel? People like to froth at the mouth and complain about vendor lock-in and Databricks and all that jazz. That's an unfair argument of half-truths that ignores the truth of the matter.

The reason that no one sees that work much, or appreciates it, is because these Kernel crates, for example, are built for a small audience: companies like ClickHouse that want to build first-class support into the Delta and Unity Lakehouse architecture.

These Kernel and Unity crates basically allow for the building of the backbone infrastructure on which the tools you and I use every day as the average data community engineer do what we want to do. We just don’t know it because others are doing the hard work of exposing these APIs for folks like DuckDB to take advantage of, and then you and I get to write a few lines of DuckDB code that connects to Delta Table and does magic.

Today, we are starting a new series in partnership with Rudderstack. Over the next several months, we'll explore modern CDPs and how they are evolving in the Age of the Agent. We'll explore both concepts and the fingers-to-the-keyboard, hard-hitting technical details, and how the face of customer data is changing as the “prompt” becomes the primary interface between user and data. Also, in the future, look forward to some great video content on CDPs in conjunction with / Early Signal.

What is a CDP and why is it important?

Many people may not have really sat down to consider what a customer data platform (CDP) is. I can say this is true for myself, even as someone who’s been in data for two decades, lived through the data upheavals from HDFS to Spark, and from SQL Server to the Lake House. I’ve seen customer records spread across various systems, both internal and external, stored in SQL Server, exported to Athena, pulled from Salesforce, combined in Tableau… and everything in between. The list of tools, sources, and types of data is as varied as the sand on the seashore, shifting and changing just as much too.

There have also been some major shifts and axioms in the data landscape at large over the last decade, including the recent Wild West of Agents; it might do us well to review and set the stage for what a CDP was, is, and could be in the future. This will be part history lesson, part speculation about the future.

The Explosion of Customer Data.

It’s no surprise to anyone in the data community that the last decade has been one full of extreme shifts in technology and business requirements in and around the technology landscape. The size, velocity, and types of data have ballooned and changed rapidly. As well, the demands from different business units regarding how data is consumed, made available, and analyzed have become complex and demanding from a technology perspective.

• Size, velocity, and types of data have ballooned

• Business and data consumption requirements have exploded

Most people who work in data grapple with these problems daily. To top it all off, customer data in particular has come to be seen as the golden key that unlocks that elusive “business value.” This pressure on customer data creates many technological demands for data teams tasked with turning complex, far-flung ideas into reality. This intersection of the “ideal” living, breathing customer dataset and the technology is where the rubber truly meets the road. At some point, Engineering and Data teams cannot hide behind an endless list of JIRA tasks for technology migrations; we must produce and enable real business value from our data so the business(es) we operate in can succeed.

• We must bring the right technology to bear to wrangle, store, serve, and make sense of a complex set of data from both internal and external systems. Oftentimes with near-real-time demands.

  • These are technical challenges

    • ex., everyone can use Kafka or Spark.

• We must do the above (technical work) with diverse business units in mind, including agentic ones that apply various critical business logic and needs.

  • These are more philosophical and business-specific.

    • e.g., everyone’s definition of what a customer lifecycle is is different.

Both the technological and business-centric demands are, each in their own right, difficult problems to solve well. They require well-thought-out technical systems architecture and design skills married to an extreme Product and Business focus that seeks to understand how the technology can meet the needs of and provide access to the (customer) data in such a way that enables the business to understand, serve, and act upon their customers in a way their competitors can’t, or don’t.

The last few decades have been the same, especially for those organizations with long histories and a complex distribution of technology bought and built to maintain and run the business. Old-school mainframes, various ERP systems of all shapes and sizes, websites and apps, databases in the cloud and some in a server closet; of course, sprinkled in is the odd SaaS platform the long line of CTOs purchased. The odd assortment of tools couldn’t be more different, and the data team tasked with bringing it all together in one spot is not to be envied. Even after that herculean task is completed, or maybe it never is finished, the real hard work has not even begun.

Various business units using these widespread tools, all with different definitions and measurements, have to be understood and information pried out, just so the odd out-of-date Dashboard can be built over and over again, leaving a long line of analytics experts crying in the wake.

If it sounds like I’ve experienced these joys, you would be right in guessing that. Customer data, in particular, is especially complex because it requires precise collection, identification, and monitoring, while business units compete to consume and use it. This doesn’t even account for the brittle, custom, legacy code and data pipelines that try to wrangle this data from its various sources into that ever-elusive “single source of truth.” Of course, combine this with the need for self-serve data, and now consumption by Agentic tooling … it’s enough to leave any CTO with a case of the shakes.

Truth be told …

• every line of custom code pulling/pushing data is a liability

• every pipeline and custom connection a breakpoint

• scattered business logic is the never-ending problem

• running a cloud data warehouse/lakehouse is a challenging problem in itself

• building customer profiles and features is rarely ever even accomplished

• integrating said customer data downstream into x, y, z marketing/sales/product etc tooling.

Of course, any astute observer will notice we conveniently left out the underlying discussion around technology … aka how the data actually flows through the system, is stored, governed, and acted upon. One could argue that historically, this was the more labor-intensive part of the process for most teams. This centers around discussion of storage solutions, databases, pipelines, code, architecture, governance, permissions; just a few pieces of the puzzle.

That being said, any good architect or staff/principal engineer who is tasked with, say, building a CDP or migration from scratch: these technology discussions are the ones that would probably keep them awake at night, mostly because they are non-negotiable pieces that have to be decided and built; otherwise, the entire stack would not operate. Think of them as table stakes: we need compute, data storage, data pipelines, governance, connections to other systems, etc., before we even get to the fun stuff of business logic and requirements. Heck, I’m sweating just writing about it.

So... what exactly is a CDP from a technical perspective?

So the more observant, technically savvy, and seasoned data professional at this point might have more questions than answers. Most likely, it wasn’t necessary to convince some of you, as I tried to convey in what we just covered, that dealing with customer data is very complicated from a business perspective, let alone adding gory details of the required implementation details like governance or storage. How can we define a Customer Data Platform? We can’t build what we don’t clearly define.

Share

A CDP is, at its core, the underlying infrastructure that a business relies on to collect, identify, model, govern, and make said data available (activate) to systems and people.

We can’t make the classic engineering mistake of getting drawn like a fly to the light where we focus on the non-additive pieces of the infrastructure and technology problem. For example, if we built the best Data Warehouse money can buy, we’d barely move the needle on the deliverables that make or break a CDP.

We don’t want to think of a CDP as simply “another database”; in fact, it is quite the opposite. What makes a CDP Platform much more than just a database with permissions is the ability within the confines of that CDP system to gather disparate data sources in a single spot, identify (who is this customer/users), model (transform, join, analyze), govern, and share said customer data. To say this another way … a CDP can answer, at a minimum …

What seems like a simple data flow on the surface can turn into truly complex technical architecture behind the scenes. We have to solve the same classic data problems from an engineering perspective, answering the question: how can we get to the desired end state with the least complexity, quickly, while delivering the desired results?

What a CDP is not (or used to be)!

I’ve experienced the pain of hand-rolled customer data solutions in the past, and it is exactly what you think it is, or can guess at. It was and is a painful process involving Data Science, Product, and Engineering, at the very least. A Data Scientist would request data from a Data Analyst, who in turn works with Engineering to access complex, stale customer data. At this point, a long process ensues where unmanaged, unreviewed SQL queries a mile long get written and rewritten.

At some point, one of the result sets is exported to Excel, passed around various groups, and then returns to its owner. Of course, what follows is even more convoluted; custom campaigns are created, passed around, and inserted into some database; questions arise about the customers, the campaigns, who has access to what, and where that nasty SQL is stored on some laptop. This is a real scenario I've experienced. What characterizes a system like this?

• not a cohesive system

• no overarching governance

• manual processes at every step

• stale data and technology

• nearly impossible to build a cohesive customer story

• external data/system integration is complex

We live in an always-on digital world where the data never sleeps. Streaming sources and events are the new norm, replacing once-dominant daily batch systems. Also, with the Lake House taking center stage in the battle for data dominance, you want CDP integrations that allow it to act as an extension of a larger Data Platform context. For example, Databricks or Snowflake integration as data sources, both in and out, is critical. Good architects and engineers look to reduce complexity and overhead while delivering real, actionable value from customer data.

Letting a Warehouse or Lakehouse do what it does well (store and make data available at scale), while enabling a CDP that can solve the complex problem of customer identity across devices and datasets; this is where the differentiator lies from both a technical perspective, as well as a business customer data activation standpoint.

Agents have changed the game.

Of course, like every other facet of business, technology, and life, the Agent revolution has upended what was previously possible in automation, analysis, and action based on real-time customer events. The fast pace of change and improvement in foundational models and agentic harnesses has both positive and negative consequences and requirements, many of which have serious implications from a CDP perspective. These agentic challenges raise real, hard-hitting questions about accuracy, observability, reliability, safety, and governance.

If a modern CDP has any chance of acting as an interface for agents, a few core tenets must be true from the beginning.

• Code-based rather than UI based interfaces.

• Machine-readable infrastructure.

• Fresh data and context.

• Deterministic customer identity resolution.

• Valid schemas, contracts, and data quality.

• Explicit governance at all levels.

• Explicit agentic boundaries.

• MCPs, APIs, CLIs

The Agentic needs are very real and growing quickly. AI provides a very real way for a modern CDP to benefit from automation, analysis, and action. But that's impossible when you just throw it on top of a system that isn’t designed to handle that complexity or use case. When we give agents access to real data, we need real boundaries around what data they can access, how, by whom, and how they can use it. Building a chatbot on top of datasets can be mildly useful, but to unlock the full potential of an agentic approach, agents must be able to take action.

Inserting Agents into the data flows gives Marketers, Product Managers, and business users the ability to build automations around previously hard-to-manage lifecycles and analysis; this takes the burden off of slow-moving, human-centric processes to pull data, analyze, create, and act. Using natural language enables business-minded power users to solve real problems, iterating faster and delivering real value.

Conclusion

At the end of the day, the need for a Customer Data Platform is not going anywhere; in fact, with the growth of AI and agentic capabilities, the need for reliable, trustworthy data has become more and more priceless. The same tools, and new ones, will continue to produce more and more customer data at an ever-increasing pace. Data quality has been and will always be an issue. Writing custom code in ten different spots to gather and wrangle source data into some form of cohesiveness will always be an architectural nightmare.

Even with AI, we can’t make the hard job of customer identity simply disappear.

Yes, the modern CDP needs to transform itself into the agentic age. Like it or not, expectations around classically slow, human workflows will no longer be tolerated as just “part of the process”; dashboards, SQL queries, spreadsheets, tickets, UIs, and manual handoffs are starting to become things of the past. Fresh data, deterministic identity, strong schemas and contracts, data quality, governance, APIs, MCPs, and explicit boundaries around what an Agent can and cannot do are no longer nice-to-haves.

Marketers, Product Managers, and other business users are under more pressure than ever to move quickly and deliver results. Data Teams are busier now than before with the added burden of agentic capabilities. Code may have gotten “cheaper,” but technical debt and the age of data problems don’t simply vanish with a new foundational model release. The truth is, the future of modern CDPs must be built on a rock-solid foundation to take advantage of what's to come.

Thanks to Rudderstack for partnering with Data Engineering Central!

The world is changing fast these days; what an exhilarating and scary time to be alive and working in tech! I love the pace at which our data community at large, both people and companies, is advancing innovation at lightning speed. Sure, we can be skeptical and somewhat angry about this Agentic Age we live in, but for good or bad, no one is going to put the genie back in the bottle, so we'd best all find our places and strap in for the ride. We should take the lemons we find and make some lemonade. I love working with all things data, as I’m sure you do, and some of the best parts of the rapidly changing landscape we find ourselves in, is the ability to do things heretofore unheard of.

To do fun things, things that may, or may not, have a point.

One feature I’ve been keeping one eye glued to lately is Databricks' Serverless compute. I don’t think it gets the discussion it should in the Lakehouse community at large, but that’s ok. Today we are going to play with it.

I’ve been a Databricks user since before it was cool, and like any good engineer, I would try to shove everything possible inside it for simplicity's sake, even if that’s not where that task should go. Part of our job is to make trade-offs, to decide which choice will tip the scale and send us down a happy path. For a long time, and maybe even still today, Databricks was and is seen as the Spark platform. But if you didn’t notice, that has changed a lot recently. In the past, I've often taken a little Python code that has nothing to do with Spark and run it inside a Databricks Job for simplicity and continuity. Should it have been running on Databricks Jobs compute? Yeah, no. But was it the best place for that “thing” to run, in the name of simplicity and oneness of a complex pipeline? Yes.

• For me, that tradeoff stopped when Serverless became an option.

When I want my architecture and pipelines to be homogeneous for numerous reasons related to ease of use, feature improvement, developer experience, simplicity … I choose Databricks because I can ship “Big Data” problems to Spark with Jobs compute, and assign lightweight actions done with tools like DuckDB, Polars, Daft, and the like … to Serverless. Fast, quick, and easy to use. Ain’t going to get any complaints from me. If the CTO comes knocking about the bill, that’s a problem for later.

Thanks to Delta for sponsoring this newsletter! I use Delta Lake daily, and I believe it represents the future of Data Engineering. Content like this would not be possible without their support. Check out their website below.

Curiosity will be the death of me.

We’ve known for some time, and I’ve written for some time about the new cadre of tools we have that can process what we used to think was Big Data on a simple, small machine. Thanks to Rust, Apache Arrow, and streaming concepts, the world is now our oyster. Why not reach for something like DuckDB or Polars inside Databricks Serverless the next time you need to munge some data? After all, both Polars and DuckDB have first-class support for Unity Catalog.

You really have no excuses, my friend.

Combine these powerful tools with Serverless compute, and you have a match made in heaven. All we are waiting for now is for you to pluck up your courage and tell Claude to commit that branch and push a PR. What are you waiting for?

For some time now, amid my busy life bustling kids to and fro, mowing the yard, interviewing folk on the podcast, and trying to keep you rabal entertained- I’ve wanted to see how DuckDB and Polars hold up to some medium-sized data, with some medium-complex workloads on Databricks Serverless pointing at a Lakehouse. I figured that on this rainy fall day, there is nothing I would rather do than try to break something. You know me.

This isn't a scientific test; if you were expecting that, you’re reading the wrong Substack. I’m curious about …

• Does Serverless compute make Databricks Spark as fast as DuckDB and Polars?

• Can we get meaningful speed improvements from DuckDB and Polars on Serverless?

• How do DuckDB and Polars deal with JOINS and Window functions?

  • aka … does real-life production complexity slow them down?

Code and Data.

Ok, so today we will be using the Backblaze open-source dataset, all about hard drive failure rates. I’ve downloaded a few years worth of data and put it into a Volume on Databricks, from which point we will ingest it into two different tables.

• First, a table with *most of the hard drive failure data

• Second, a table with metadata about each specific hard drive model

  • This will be used for the Join

This is a classic Spark problem: one large table joining to a very small one, aka a broadcast join. I’m curious how well DuckDB and Polars handle this scenario. Also, we will apply a Window function of some sort to the entire dataset, as if we were doing some analysis. We will run the same query on DuckDB, Polars, and Databricks Spark using Serverless.

• We will be working with about 285,339,435 rows.

from pyspark.sql.types import (
    StructType,
    StructField,
    StringType,
    DateType,
    LongType,
    IntegerType
)

smart_ids = [
    1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13, 15, 16, 17, 18,
    22, 23, 24, 27, 71, 82, 90,
    160, 161, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172,
    173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184,
    187, 188, 189, 190, 191, 192, 193, 194, 195, 196, 197, 198,
    199, 200, 201, 202, 206, 210, 218, 220, 222, 223, 224, 225,
    226, 230, 231, 232, 233, 234, 235, 240, 241, 242, 244, 245,
    246, 247, 248, 250, 251, 252, 254, 255
]

schema = StructType([
    StructField("date", DateType(), True),
    StructField("serial_number", StringType(), True),
    StructField("model", StringType(), True),
    StructField("capacity_bytes", LongType(), True),
    StructField("failure", IntegerType(), True),
    StructField("datacenter", StringType(), True),
    StructField("cluster_id", StringType(), True),
    StructField("vault_id", StringType(), True),
    StructField("pod_id", StringType(), True),
    StructField("pod_slot_num", IntegerType(), True),
    StructField("is_legacy_format", IntegerType(), True),
])

for smart_id in smart_ids:
    schema.add(
        StructField(
            f"smart_{smart_id}_normalized",
            IntegerType(),
            True
        )
    )
    
    schema.add(
        StructField(
            f"smart_{smart_id}_raw",
            LongType(),
            True
        )
    )

df = (
    spark.read
        .option("header", "true")
        .option("dateFormat", "yyyy-MM-dd")
        .schema(schema)
        .csv("dbfs:/Volumes/workspace/default/test-data/*/*")
)
df.write.saveAsTable('raw_data')

df = df.select('serial_number', 'model').distinct()
df.write.saveAsTable('models')

We can use this code to write the two Delta tables.

Ok, I know I have the model in both datasets, but this is just for fun; we want the join and some window function that we can run across Polars, DuckDB, and Databricks Spark in our Notebook attached to Serverless. See what happens, any guesses? Don’t cheat.

Code

Here we go with PySpark, nothing fancy here, just some kinda this-and-that you would expect to see in a production pipeline were some analytics are being calculated off some bronze layer and shoved into some report for the powers that be.

from pyspark.sql import functions as F
from pyspark.sql.window import Window
import time

raw = spark.table("raw_data")
models = (
    spark.table("models")
    .select("serial_number", "model")
    .dropDuplicates(["serial_number"])
)

ordered_window = (
    Window
    .partitionBy("serial_number")
    .orderBy("date")
)

rolling_window = (
    Window
    .partitionBy("serial_number")
    .orderBy("date")
    .rowsBetween(-29, 0)
)

start = time.perf_counter()

result = (
    raw
    .drop("model")
    .withColumn(
        "row_num",
        F.row_number().over(ordered_window)
    )
    .withColumn(
        "previous_smart_5_raw",
        F.lag("smart_5_raw", 1).over(ordered_window)
    )
    .withColumn(
        "rolling_avg_smart_5_raw",
        F.avg("smart_5_raw").over(rolling_window)
    )
    .join(
        models,
        on="serial_number",
        how="left"
    )
)

benchmark_result = (
    result
    .agg(
        F.count("*").alias("row_count"),
        F.sum("row_num").alias("sum_row_num"),
        F.sum("previous_smart_5_raw").alias("sum_previous"),
        F.sum("rolling_avg_smart_5_raw").alias("sum_rolling_avg")
    )
    .collect()
)

elapsed = time.perf_counter() - start

print(benchmark_result)
print(f"PySpark elapsed: {elapsed:.2f} seconds")

DuckDB code.

Ok, so let’s move to DuckDB. This is the code that will run outside the timer, just setting up a connection from DuckDB to Unity Catalog. DuckDB provides a nice integration into Unity Catalog would should make our lives easier.

I don’t wan you to just slide past this without thinking. In the varied and Lakehouse world we live in, it’s extremely important that our tools talk to each other, otherwise things become an impossible mess of cobbled together solutions that cause more bugs and complexity, providing not enough bang-for-the-buck.

DuckDB code.

import duckdb
import time

con = duckdb.connect()

con.execute("""
INSTALL unity_catalog;
LOAD unity_catalog;
""")

con.execute(f"""
CREATE SECRET (
    TYPE unity_catalog,
    TOKEN 'my_token',
    ENDPOINT 'https://my-workspace.cloud.databricks.com/'
);
""")

con.execute("""
ATTACH 'confessions' AS uc (
    TYPE unity_catalog,
    DEFAULT_SCHEMA 'default'
);
""")

And then the actual code to run. You will notice this code isn’t any more complex or verbose than the PySpark code, so we aren’t really giving up much switching to Databricks, it will be very interesting to see the runtimes will it not?

The big idea here is to find out if we can combine the power of Serverless … that quick hitting compute, with DuckDB … known to be fast … and start to build out more interesting workflows inside Databricks. One’s that are fast, and save time and money.

query = """
WITH models_deduped AS (
    SELECT
        serial_number,
        ANY_VALUE(model) AS model
    FROM uc.default.models
    GROUP BY serial_number
),

windowed AS (
    SELECT
        r.* EXCLUDE (model),

        ROW_NUMBER() OVER (
            PARTITION BY serial_number
            ORDER BY date
        ) AS row_num,

        LAG(smart_5_raw, 1) OVER (
            PARTITION BY serial_number
            ORDER BY date
        ) AS previous_smart_5_raw,

        AVG(smart_5_raw) OVER (
            PARTITION BY serial_number
            ORDER BY date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS rolling_avg_smart_5_raw

    FROM uc.default.raw_data r
),

joined AS (
    SELECT
        w.*,
        m.model
    FROM windowed w
    LEFT JOIN models_deduped m
        ON w.serial_number = m.serial_number
)

SELECT
    COUNT(*) AS row_count,
    SUM(row_num) AS sum_row_num,
    SUM(previous_smart_5_raw) AS sum_previous,
    SUM(rolling_avg_smart_5_raw) AS sum_rolling_avg
FROM joined
"""

start = time.perf_counter()

benchmark_result = con.execute(query).fetchall()
elapsed = time.perf_counter() - start

print(benchmark_result)
print(f"DuckDB elapsed: {elapsed:.2f} seconds")

Oh so close. Always’ something ain’t it?

There appears to be some permission problems where Databricks/Unity only wans to give READ permission to DuckDB, but it automatically wants to have READ/WRITE, so it gets rejected out of the box, even though my query only is SELECT. Is this a problem with Databricks Serverless because it’s using Spark Connect API, so Databricks sees DuckDB as not a “Databricks Compute Envrionment?” Poohy. I guess our option here is to recreate these tables as “External Managed tables” with a custom s3 URI. Stupid, because that’s technically what they already are since this is a Databricks + AWS setup I have!

I think I found a related open issue for this DuckDB + Unity integration.

I dropped the managed tables, and re-ran this.

from pyspark.sql.types import (
    StructType,
    StructField,
    StringType,
    DateType,
    LongType,
    IntegerType
)

smart_ids = [
    1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13, 15, 16, 17, 18,
    22, 23, 24, 27, 71, 82, 90,
    160, 161, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172,
    173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184,
    187, 188, 189, 190, 191, 192, 193, 194, 195, 196, 197, 198,
    199, 200, 201, 202, 206, 210, 218, 220, 222, 223, 224, 225,
    226, 230, 231, 232, 233, 234, 235, 240, 241, 242, 244, 245,
    246, 247, 248, 250, 251, 252, 254, 255
]

schema = StructType([
    StructField("date", DateType(), True),
    StructField("serial_number", StringType(), True),
    StructField("model", StringType(), True),
    StructField("capacity_bytes", LongType(), True),
    StructField("failure", IntegerType(), True),
    StructField("datacenter", StringType(), True),
    StructField("cluster_id", StringType(), True),
    StructField("vault_id", StringType(), True),
    StructField("pod_id", StringType(), True),
    StructField("pod_slot_num", IntegerType(), True),
    StructField("is_legacy_format", IntegerType(), True),
])

for smart_id in smart_ids:
    schema.add(
        StructField(
            f"smart_{smart_id}_normalized",
            IntegerType(),
            True
        )
    )

    schema.add(
        StructField(
            f"smart_{smart_id}_raw",
            LongType(),
            True
        )
    )


RAW_PATH = (
    "s3://confessions-of-a-data-guy/"
    "benchmarks/raw_data"
)

MODELS_PATH = (
    "s3://confessions-of-a-data-guy/"
    "benchmarks/models"
)

RAW_TABLE = "confessions.default.raw_data"
MODELS_TABLE = "confessions.default.models"

raw_df = (
    spark.read
        .option("header", "true")
        .option("dateFormat", "yyyy-MM-dd")
        .schema(schema)
        .csv("dbfs:/Volumes/confessions/default/test-data/*/*")
)

models_df = (
    raw_df
        .select(
            "serial_number",
            "model"
        )
        .distinct()
)

spark.sql(
    f"DROP TABLE IF EXISTS {RAW_TABLE}"
)

spark.sql(
    f"DROP TABLE IF EXISTS {MODELS_TABLE}"
)

(
    raw_df.write
        .format("delta")
        .mode("overwrite")
        .option("path", RAW_PATH)
        .saveAsTable(RAW_TABLE)
)

(
    models_df.write
        .format("delta")
        .mode("overwrite")
        .option("path", MODELS_PATH)
        .saveAsTable(MODELS_TABLE)
)

This is a serious problem, because it means all those wondeful Managed Unity tables you have can’t be touched by external engines on Serverless … meaning this benchmark is pointless because no one is going to recreate all their tables as EXTERNAL unless they have a screw loose.

Look, it’s entirely possible I’m missing something obvious, maybe there is another obvious workround that I should know about and try. But there are only so many hours in the day.

Ok, after recreating tables as EXTERNAL my woes continue.

After I fixed this error, doing some GRANTS etc, I finally got this error.

I just don’t think that Serverless is ready for DuckDB time. Well, I’ve put enough time into this I don’t want to abandon what we’ve been working on so far, we gotta finish one way or the other. I guess we could switch over to Jobs Compute and see what happens there. I’m very curious if Polars has the same problems with Serverless.

Polars on Serverless.

Ok, let’s switch over to Polars for Unity Catalog, and do the same benchmark, maybe their Databricks/Unity are better than DuckDB. I’ve always had my troubles in the past with Polars and anything approaching a “large” dataset. They have been plauged for years by OOM issues that are random and often hidden under the rugs. If you bring them up, the Dark Lord will come find you.

import polars as pl

WORKSPACE_URL = "https://dbc-9a64f31c-25b9.cloud.databricks.com/"

catalog = pl.Catalog(
    WORKSPACE_URL,
    bearer_token="blahblah"
)

… and the code.

raw = catalog.scan_table(
    "confessions",
    "default",
    "raw_data"
)

models = (
    catalog
    .scan_table(
        "confessions",
        "default",
        "models"
    )
    .select([
        "serial_number",
        "model"
    ])
    .unique(
        subset=["serial_number"]
    )
)

start = time.perf_counter()

result = (
    raw
    .drop("model")
    .sort([
        "serial_number",
        "date"
    ])

    .with_columns([

        pl.col("date")
        .cum_count()
        .over("serial_number")
        .alias("row_num"),

        pl.col("smart_5_raw")
        .shift(1)
        .over("serial_number")
        .alias("previous_smart_5_raw"),

        pl.col("smart_5_raw")
        .rolling_mean(
            window_size=30,
            min_samples=1
        )
        .over("serial_number")
        .alias("rolling_avg_smart_5_raw")
    ])

    .join(
        models,
        on="serial_number",
        how="left"
    )

    .select([
        pl.len().alias("row_count"),

        pl.col("row_num")
        .sum()
        .alias("sum_row_num"),

        pl.col("previous_smart_5_raw")
        .sum()
        .alias("sum_previous"),

        pl.col("rolling_avg_smart_5_raw")
        .sum()
        .alias("sum_rolling_avg")
    ])
)

benchmark_result = result.collect()

elapsed = time.perf_counter() - start

print(benchmark_result)

print(
    f"Polars elapsed: {elapsed:.2f} seconds"
)

Polars ran for about 8 minutes and then everything just died. Not sure why, why it died, and why it would take so long to run the query. Anywho, no surprise there. I honestly wasn’t expecting it to run, although not having it puke on the Unity Catalog connection was a nice surprise. Got closer than DuckDB did.

Back to DuckDB. We ain’t giving up yet.

Not sure why I didn’t think of this off the bat, but since I can get the s3 location for our Delta Lake tables in s3, what if I just give DuckDB my AWS creds and the s3 table locations, so leave the code mostly the same, will it run? Maybe this whole Unity thing and DuckDB just don’t jive well.

This is what our new and updated DuckDB code looks like. Now, I’m not happy about having to do this, I mean in theory any Databricks + Unity setup on AWS we would be able to find even for Managed tables that s3 location and feed it in, but it feels a little hacky and not something we would want to do long term in a large codebase with tons of tables.

con = duckdb.connect()

con.execute("""
INSTALL delta;
LOAD delta;

INSTALL aws;
LOAD aws;
""")
con.execute(f"""
CREATE OR REPLACE SECRET s3_secret (
    TYPE s3,
    KEY_ID 'blah',
    SECRET 'blahblahblah',
    REGION 'us-east-1'
);
""")

… and the benchmark code.

query = """
WITH models_deduped AS (
    SELECT
        serial_number,
        ANY_VALUE(model) AS model
    FROM delta_scan(
    's3://confessions-of-a-data-guy/benchmarks/models'
)
    GROUP BY serial_number
),

windowed AS (
    SELECT
        r.* EXCLUDE (model),

        ROW_NUMBER() OVER (
            PARTITION BY serial_number
            ORDER BY date
        ) AS row_num,

        LAG(smart_5_raw, 1) OVER (
            PARTITION BY serial_number
            ORDER BY date
        ) AS previous_smart_5_raw,

        AVG(smart_5_raw) OVER (
            PARTITION BY serial_number
            ORDER BY date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS rolling_avg_smart_5_raw

    FROM delta_scan(
    's3://confessions-of-a-data-guy/benchmarks/raw_data'
) r
),

joined AS (
    SELECT
        w.*,
        m.model
    FROM windowed w
    LEFT JOIN models_deduped m
        ON w.serial_number = m.serial_number
)

SELECT
    COUNT(*) AS row_count,
    SUM(row_num) AS sum_row_num,
    SUM(previous_smart_5_raw) AS sum_previous,
    SUM(rolling_avg_smart_5_raw) AS sum_rolling_avg
FROM joined
"""

start = time.perf_counter()

benchmark_result = con.execute(query).fetchall()

elapsed = time.perf_counter() - start

print(benchmark_result)
print(f"DuckDB elapsed: {elapsed:.2f} seconds")

DuckDB was slower than I expected, but it finished at least!

Funny we ended up here uh?! Isn’t this just exactly how engineering things works when we are working with new technologies? It never really goes how we think it’s going to go, but we must beat or way through the weeds until we get to our endpoint.

• DuckDB with Unity Catalog pukes, but at least runs if we bypass Unity.

• Polars just pukes can’t finish.

Like any good Polars bug that involves memory, from the beginning of time this is the way, to close the issue on GitHub and pretend like it’s not a think.

I figured I would try my best to give Polars the best change possible, maybe add some column selections to ensure we only have the data we want, not assume any predicate pushdowns or whatever.

import polars as pl
import time

raw = (
    catalog
    .scan_table(
        "confessions",
        "default",
        "raw_data"
    )
    .select([
        "serial_number",
        "date",
        "smart_5_raw"
    ])
)


models = (
    catalog
    .scan_table(
        "confessions",
        "default",
        "models"
    )
    .select([
        "serial_number",
        "model"
    ])
    .unique(
        subset=["serial_number"]
    )
)

start = time.perf_counter()


result = (
    raw
    .sort([
        "serial_number",
        "date"
    ])

    .with_columns([
        (
            pl.int_range(pl.len())
            .over("serial_number")
            + 1
        )
        .alias("row_num"),

        pl.col("smart_5_raw")
        .shift(1)
        .over("serial_number")
        .alias("previous_smart_5_raw"),

        pl.col("smart_5_raw")
        .rolling_mean(
            window_size=30,
            min_samples=1
        )
        .over("serial_number")
        .alias("rolling_avg_smart_5_raw")
    ])
    .join(
        models,
        on="serial_number",
        how="left"
    )
    .select([
        pl.len().alias("row_count"),

        pl.col("row_num")
        .sum()
        .alias("sum_row_num"),

        pl.col("previous_smart_5_raw")
        .sum()
        .alias("sum_previous"),

        pl.col("rolling_avg_smart_5_raw")
        .sum()
        .alias("sum_rolling_avg")
    ])
)

benchmark_result = result.collect()
elapsed = time.perf_counter() - start

print(benchmark_result)
print(f"Polars elapsed: {elapsed:.2f} seconds")

No such luck, same error for Polars. Probably just more classic memory errors.

The cluster was unhealthy or has been terminated. Please restart the cluster 
or attach this notebook to a different cluster.

Ok, one last try on Polars, one little update. Come on, you can do it!

benchmark_result = result.collect(
    engine="streaming"
)

Fatal error: The Python kernel is unresponsive.
---------------------------------------------------------------------------
The Python process exited with exit code 137 (SIGKILL: Killed). This may have been caused by 
an OOM error. Check your command's memory usage.

I tried my friends, this is no surprise to me, Polars has OOM demons that are secret and must never ever be named outloud lest the Dark Lord come find you.

Moral of the story.

This was an interesting one and not what I expected. In my mind it made perfect sense that both Polars and DuckDB would made for running on Serverless … both tools are usually fast, both in theory have Unity Catalog support; it just seemed to make logical sense that we could save compute and runtimes by swapping in DuckDB or Polars for typical PySpark pipelines … I could have not been more wrong. I should have known.

• DuckDB was way slower than expected on remote Delta Lake tables.

• Polars in classic fashion, as always, puked on memory.

• DuckDB doesn’t play with Unity Catalog via Serverless.

• PySpark on Serverless apparently is fastest.

This is why even with agents writing all our code, there is still room for us humans who like to think, poke, prod, and test new things. I’ve always felt that AI isn’t very creative, it’s been trained on the past, what good is it for pushing the edges of what we can do to in the future?

This shows me there is still room for humans in the mix.

Humans that try to do things that haven’t been done before, break things, try things, be creative with things, push the boundaries of things, try new ways of solving age old problems. Data is messy, making tools place nice with each other is messy. There are still new problems to solve!

Come join me over on YouTube if you get bored!

I’ve been asked a few times about my thoughts on the now-old news of AWS acquiring DuckDB Labs. Initially, I did blast some thoughts out on my old blog, who knows what the truth really is, not I. But it did send a few shockwaves initially through the data community at large, bringing to the surface long-stoked fears that the future of open source is bowing before the knees of billion-dollar corporations. Such is life; not much has changed over the eons. Today I wanted to give you a different perspective on why AWS acquired DuckDB Labs … and not Polars. It’s a sordid tale that mostly involves what I would call “developer goodwill.”

One could generally argue that the world we live on, this spinning rock populated by eccentric humans, seems to follow a set of rules, with some appearance of consistency. Namely, that age old tale of good vs evil … call it karma if you like.

This little vignette of data products playing out before our eyes like some Odyssey, can teach us important life lessons.

Read more

In this episode of Data Engineering Central, I sit down with Maria Vechtomova, AI Engineering Lead, co-founder of CAUCHY, Databricks MVP, and author of MLOps with Databricks.

Maria has spent more than 12 years working across data, machine learning, MLOps, and AI engineering. We talk about how production ML has evolved from the early days of homegrown model platforms to today’s world of LLMs, agents, Databricks, MLflow, and AI-assisted coding.

We dig into why LLM applications can be significantly harder to evaluate and monitor than traditional ML, the growing importance of tracing and observability, and why governance becomes more complicated when agents can interact with data and other systems. Maria also explains why many of the problems we’re facing with AI aren’t actually new — AI has simply made them much harder to ignore.

• We also get into AI coding tools, testing AI-generated code, Databricks and Unity Catalog, the rapidly changing LLMOps ecosystem, and what engineers should focus on as AI changes the way software is built.

One of my favorite ideas from the conversation: AI is a mirror. It can help you write a lot more code, a lot faster — but whether that code is actually good still depends on the engineer behind it.

Maria also shares what she thinks matters most for engineers entering the field today: common sense, soft skills, and business sense.

Topics include: MLOps and LLMOps, production AI, Databricks and MLflow, AI agents, observability and tracing, model and LLM evaluation, AI governance, Unity Catalog, AI-generated code, testing, and the future of AI engineering.

TDLR: Databricks’ new FILE type brings unstructured assets (multimodal data) ((video, audio, etc)) into the Lakehouse with the ease that we expect in this chaotic and agentic age we find ourselves in. I’ve been thinking more and more about multimodal data ever since Tyler brought it to my attention early on. On top of that, finding myself doing more and more unstructured data things these days tends to change how one thinks, such as building agentic solutions at the $day_job. One has to imagine that this type of data will only increase in popularity and, in fact, become more of a requirement for many data engineers in the future. By “this”, I mean munging unstructured or untraditional data into the Lakehouse that was once ruled by tabular data.

If your curious about an introduction to multimodal data, or what it even is, checkout this previous blog post.

Pontificating on concepts and architecture.

Ok, before you all start to berate me about how the new FILE data type from Databricks is no big deal, I implore you to soothe your inner angry engineer and bring forth your kind systems architect inner child. We need to sometimes zoom out a little and broaden our horizons, measure things by a different ruler than what we are used to. For as long as I’ve been hacking out code, it’s always been true that programmers have a certain foible to laud the complex, hard, and technically impressive “things.” Now, this isn’t bad in and of itself. But it’s the same reason many a “smart” engineer languishes in a dark cubicle while their “less smart” peer continues to advance.

This is that sort of innovation, simple but useful. Just because something appears simple on the surface, doesn’t mean it should be simply ignored. Why should we care about FILE data types on Databricks Lakehouse?

• Bringing the management of ALL data into the Lakehouse.

• Bringing logic and data processing into a single spot.

• Architectural simplicity and reducing complexity.

• Unity Catalog governance on top of unstructured data.

Simple changes in architecture and systems design can, and do, breed all sorts good, or bad things. Doing things in separate systems always creates overhead and technical challenges that suck the lifeblood out of innovation. Having a single source system that controls ALL the data, no matter the kind, allows for fast and rapid product and engineering iteration. This is how we should view the new FILE data type being provided on the Databricks platform. It gives a reason to bring previously unstructured data that we didn’t view as “fit for the lakehouse” into the fold.

As designers of data platforms and systems, we should embrace the ability to hack off other systems that introduce bottlenecks to our ulitmate goals.

Lets talk about the FILE data type.

Ok, so let’s get down the brass tacks. At some point we can test out this new FILE data type in Databricks but storing some video shorts from my podcast, which you had better be very familiar with. But, for now let’s make what we can out of the technical details of a FILE data type.

Again, of course the more maniacal among us might think, “So, we could just store a file URI as a STRING and dump bytes into some column. Well, yes, I guess you could do that; it’s also one of those ideas that sounds great in theory but would fail in production with all its unthought-of glory. Again, anyone can do anything with enough ‘umph, but that doesn’t mean you should.

We can think of a FILE value in two different parts.

• its metadata.

  • uri, size, content_type, and checksum

• its content.

  • FILE EXTERNAL or FILE MANAGED

The last concept is probably one of the more important when it comes to FILE data types, deciding if FILE EXTERNAL fits your needs, or FILE MANAGED. If you are at all familiar with Databricks you’ve probably run across this sort of “managed” vs “unmanaged” concepts with Tables , Delta Lake style. We should always do everything possible to ensure we use the MANAGED options so we get the ease of built-in permissions, governance, and the cleanup of unreferenced or unused data. Otherwise, you are left to manage the details yourself, which will promptly be ignored or forgotten.

If indeed you are smart, as I know you are, picking the FILE MANAGED option is what you will do. But, this brings up another concept you will have to think about called FileSpace . This is a simple and obvious concept: a FileSpace is a Unity Catalog Volume that you will use for the table, with the File data type, that will hold the actual files themselves. This is needed because when we have “managed” File data, and say we delete a file … it can be garbage collected later and cleaned up.

Playing with the new FILE data type.

Ok, so let’s actually put our hands to the plow and do a little thing. See if the fuss around this new data type lives up to the hype I’ve given it. We will do a few straightforward things to see how helpful, in the real world, it is to work with this ability to play with non-conventional data in the Lakehouse world that’s always trying to keep us down. Here is my plan.

• Create a new managed Volume

• Create a new Lakehouse table (Delta) with FILE column.

  • which uses the above Volume

• Ingest some video data (shorts) from my Podcast.

• Try to write a Python UDF to work on this data.

  • Let’s try to transcribe the video to text and store these STRINGs.

This seems like a totally reasonable thing to do, to touch what we need to and make sure it all meets our expectations.

First, let’s create a Volume that can back our soon-to-be existing Delta Lake table with a File column, backed by this Volume we just created.

Next, we can right the simple DDL required to create a table to hold my Podcast data, including the FILE type, using the new Volume as the store for this data.

Of course, this was the easy part; first things first, can we ingest a video file from S3 into our newly created table? I suggest that if you want a deeper dive into how to ingest different types of data into these new tables that have a FILE type that could be managed or unmanaged, that you read this documentation.

I don’t feel like diving too deeply into this yet, because I don’t know what I’m talking about, but where and how in a production system would you land such unstructured data files? One would have to assume either shipping files straight into a Databricks Volume, or more likely, landing them in S3 or some other cloud storage. This question, in and of itself, begs a deeper dive into what a production ingestion pipeline would look like. We wouldn’t want to store large files in two places, methinks. I love rabbit holes, but I must avoid this one for now.

Back to the issue at hand. I now have 3 mp4 files sitting in S3 at s3://confessions-of-a-data-guy/video_data/ , let’s write some Python in a Databricks Notebook and see what it’s lake to get some actual data into our new Delta table with FILE data types!

Let’s write a simple PySpark script for our first attempt at ingesting this data.

from pyspark.sql import functions as F

SOURCE_PATH = "s3://confessions-of-a-data-guy/video_data/"
TABLE = "podcasting"

videos = (
    spark.read
    .format("file")
    .load(SOURCE_PATH)
    .select(
        F.xxhash64(F.col("path")).cast("bigint").alias("id"),
        F.regexp_replace(
            F.col("_metadata.file_name"),
            r"\.[^.]+$",
            "",
        ).alias("title"),

        # Populate later with the Whisper transcription UDF.
        F.lit(None).cast("string").alias("transcript"),
        F.col("file"),
    )
)

videos.writeTo(TABLE).append()

This is about as straightforward code as it comes; of course, in a production scenario, we might have a Lakeflow pipeline or something else adding new files as they arrive. And we hit our first error. studio.mp4 is not a supported ingestion source for managed file type

SparkException: Job aborted due to stage failure: java.lang.SecurityException: Source path s3://confessions-of-a-data-guy/video_data/riverside_common_sense in engineering_dan_beach studio.mp4 is not a supported ingestion source for managed file type.
	at com.databricks.sql.fileref.UploadToManagedStorage.$anonfun$validateSourcePath$1(FileReferenceCreationModes.scala:116)
	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)

Of course, trusty AI helped me understand what I did wrong, which makes sense, but is also pretty annoying.

That error occurs because Databricks cannot ingest a  FILE MANAGED  value directly from a raw  s3://  path. Stage the MP4s into a separate Unity Catalog volume first, then read the staged files as  FILE  references and append them to  podcasting . Do not use the  data-engineering-central-files  FileSpace volume as staging; 
it is reserved for the table’s managed files.

Here is my complaint: what if I have tons of pre-existing video/and other non-traditional data sitting in cloud storage somewhere? This could be a serious problem in a large production system with unstructured audio and video data; we could have many hundreds of TBs of data that would have to be transferred into a Volume. This could be a showstopper for large orgs. I also, when reading the docs, missed a critical note.

But this takes us back to Volumes , could we just do an EXTERNAL Volume pointed to our/my S3 bucket?

CREATE EXTERNAL VOLUME confessions.default.raw_podcast_data
LOCATION 's3://confessions-of-a-data-guy/video_data/';

Let’s rework our code a little bit then to match this new layout.

from pyspark.sql import functions as F

SOURCE_VOLUME_PATH = "/Volumes/confessions/default/raw_podcast_data/"
TABLE = "podcasting"

source_videos = (
    spark.read
    .format("file")
    .load(SOURCE_VOLUME_PATH)
    .filter(F.lower(F.col("path")).rlike(r".*\.mp4$"))
    .select(
        F.xxhash64(F.col("path")).cast("bigint").alias("id"),
        F.regexp_replace(
            F.col("_metadata.file_name"),
            r"\.[^.]+$",
            "",
        ).alias("title"),
        F.lit(None).cast("string").alias("transcript"),
        F.col("file"),
    )
)

new_videos = source_videos.join(
    spark.table(TABLE).select("id"),
    on="id",
    how="left_anti",
)

new_videos.writeTo(TABLE).append()

Worked like a charm!

And just like that we have ingested our unstructured data, video in this case, right into our Lakehouse, with all the power that comes with such a move! It was pretty simple to do, and just like that we now have best-in-class security, governance, and the power of the Databricks platform at our fingertips, with this once hard-to-manage data ready to be told what to do!

Amazing!

Of course, we would probably have that Python UDF working to start with and run it in the same script, but for fun, let’s see if we can get something written that will read the files, get the transcript using Whisper, and then MERGE back into the table. We have solved the problems and tried out the FILE type, but let’s keep kicking it a little.

Share

from pyspark.sql.types import FileRef, StringType

_model_cache = {}


def transcribe_video(file: FileRef) -> str:
    if file is None:
        return None

    import os
    from pathlib import Path

    import imageio_ffmpeg
    import torch
    import whisper

    if "whisper" not in _model_cache:
        _model_cache["whisper"] = whisper.load_model("tiny.en", download_root="/tmp/whisper")

    ffmpeg_dir = str(Path(imageio_ffmpeg.get_ffmpeg_exe()).parent)
    os.environ["PATH"] = f"{ffmpeg_dir}:{os.environ['PATH']}"

    result = _model_cache["whisper"].transcribe(
        str(file.as_local_file()),
        language="en",
        fp16=torch.cuda.is_available(),
        verbose=False,
    )

    return result["text"].strip()


spark.udf.register(
    "transcribe_video",
    transcribe_video,
    StringType(),
)
spark.sql("""
MERGE INTO podcasting AS target
USING (
  SELECT
    id,
    transcribe_video(file) AS transcript
  FROM podcasting
  WHERE transcript IS NULL
) AS source
ON target.id = source.id
WHEN MATCHED THEN
  UPDATE SET target.transcript = source.transcript;
  """)

Hotdog! Look at that, ma! She can print this off and tape it up on the fridge.

So, here’s the deal: simply storing some mp4 files in cloud storage, reading them with Python, and ripping out the transcript isn't amazing or groundbreaking. I urge you to put your inner systems-architect hat on for a minute and set down the engineer hat. It’s about WHERE this is all taking place, and what that means at a 10,000-foot level.

This is happening inside Unity Catalog, with all that this entails at an enterprise level.

First-class data citizens are being created from data we typically wouldn't find in most Lakehouse architectures. This is a game changer for orgs wanting to bring the power of Unity Catalog and Databricks to bear on ALL data types. PDFs, Docs, Video, Audio, or whatever, doesn’t matter. This is how innovation happens; this is how costs and complexity that have been around for decades are deleted in a single swoop. I’m sure this is only the beginning of what Databricks has planned in the multi-modal data space, we are sure to be on the edge of some exciting features and improvements in the future!

Leave a comment

Come join me over on YouTube if you get bored!

To be honest I’ve long ago given up on the hectic page of changes in the AI world, there’s just too much for one poor guy, stuck in the middle of a cornfield, to track. The marketing hype train about the newest Foundational Model turned out to be the little boy who cried wolf. I suppose it’s both good and bad for the power users out there, the folk on the ground trying to deploy production ready solutions. There is always something new and improved, but it’s extremely difficult to find ones self on solid footing where we feel sure of ourselves. One could argue that this is where the fun is found though, on the rough edges of the goings and comings of features.

In the $day_job I’ve had the pleasure of building a few different “agentic applications,” from chatbots to more traditional question and answer pipelines. So, although I’m not Databricks AI expert by any stretch of the imagination, I do have some thoughts on generally what it looks like to serve LLM traffic via Databricks; a platform I would argue that makes it so simple, it’s criminal. I want to explain why that is. I also want to talk at a 10,000 foot level what you should be thinking about when starting an LLM/Agentic project of some sort. Hopefully what I’ve learned will help those of you who might be new or just starting down that long and winding road.

This is NOT a post about “how do I setup a provisioned model serving AI Gateway LLM on Databricks.” This is more of a conceptual point of view of …

• What are my options?

• How should I generally think about Databricks LLM options?

• What ancillary features affect my decision?

The truth is the answer is always and forever, “it depends.” It depends on your content, business context, cost context, engineering requirements, etc. It always depends. I’m just giving you my take based on my context, that might be different for you.

Thanks to Delta for sponsoring this newsletter! I use Delta Lake daily, and I believe it represents the future of Data Engineering. Content like this would not be possible without their support. Check out their website below.

Yeah, so … what are my options?

Ok, I’m pretty sure I put enough caveats in place, but in case I didn’t … this stuff is changing rapidly all the time, what I say might be changed, depreciated, or replaced by the time you finish reading. Use at your own risk. I’m going to start under the assumption that you are coming in fresh and clean as a whistle, your boss says “We need this fairly complex application built in the next few weeks, use our Databricks account … see ya later.” Again, who knows what you’re building, but you need an LLM to call, what next?

So, right out the box you have to understand a general boxes, or flavors of LLM serving. Each one fits a different use case, of course with plenty of overlap. Let’s draw a picture first. Yeah, yeah, I’m oversimplifying the matter, but that is my intention. I want to hide some of the nuance and think generally about what buckets we can put Databricks LLM/Agentic offerings in. Sure, we have to draw the lines somewhere; yes they are blurred, yes they change a lot, but we have to draw them at some point.

I went ahead and gave you my answers (When should I use what?) a little early, but we will unpack my reasoning a little more. For now, at the time of writing, I see the options as follows from a very simplistic point of view.

• pay-per-token

  • the serverless of LLM APIs

• provisioned throughput

  • the ultimate in LLM infererence options

• external models

  • for massive teams wanting control of Anthropic/OpenAI etc.

• custom model/Agents serving

  • the 1% doing Lord knows what

For todays discussion, I’m more concerned with talking about the two I used most, and I’m guessing a lot of others will use a lot. Yes, I’ve done the Databricks App route with their button-click/terraform easy deployment of an agent, indeed it is nice. But, that topic is due it’s own post, so I’m going to skip that customer LLM/Agent serving product. I’m also only going to only touch briefly on external models, because that is a specific audience as well. Instead, I will focus on two popular options, pay-per-token Foundational Models and Provisioned Throughput. These two will keep us busy enough as it is.

Data Engineering Central is a reader-supported publication. To receive new posts and support my work, consider becoming a free or paid subscriber.

What do you need besides an LLM Endpoint?

Sure, so if you’re like me, half caveman, you always reach for the simplest options and lowest hanging fruit as the panacea to all engineering evils. This approach can sometimes come back to bite you in the old rear-end if you’re not careful. When it comes to AI, being the new Spice of Universe that is very new, the requirements and what the business requires can be a little complex. I would suggest that when choosing a product to serve some LLM use case, that you think more than just about “I need an API for inference.” That is much too simplistic and misses a lot of other requirements that you will have.

Here are some areas of immediate concern that you will need to work through.

Don’t get me wrong, there are a lot of things to worry about when throwing an API up into the cloud for others to use. Just about everything can, and will, go wrong if you aren’t prepared for it. Could be the users, could be the questions from your CTO; either of which can put you in an awkward position. When a Data Scientist gets excited and blows the token budget, or someone asks in a meeting where the logs are stored so they can see what users are doing and you start to cough … or worse yet, the CTO says “Since I’m sure not everyone can access this model, which Group did you give permissions too?” Heck, maybe it will be the ole’ cost question, so based on X usage, what do you think the bills going to be?

• Logging

  • who’s doing what, audit

• Monitoring

  • tracking spend and usage

• Governance

  • who can access what, rate limits, routing, etc

It’s hard to dig into every single detail, but at the very least you should be concerned and consider logging, monitoring, and governance. Of course wrapped up in all that is important topics, cost, rate limits, and all the rest. To top it all off, we can’t really talk about these topics without talking about Unity Gateway.

Unity Gateway as the entrypoint.

What is Unity Gateway? Think of the LLM products we talked about above, like Foundational Models via API per token, or Provisioned Throughput as the the serving mechanism and Unity Gateway as the “how you govern and manage that served model.” And yes my friend, indeed you do want to govern your LLM. An unfettered LLM API is about as dangerous as it comes. You wouldn’t leave any other API built over the last decade just blowing in the breeze, so why would an LLM API being served be any different? It’s not.

We care about who can do what, what’s happening, cost and controls … this is what Unity Gateway offers. The same first class controls we are used to on other Unity governed objects, applied to our AI tools.

Look, it’s hard enough to build this agentic products, let along hand-rolling some bespoke monitoring solution on top of it all. We get paid to deliver results, not reinvent the wheel.

Observe and Monitor

For example, you need to understand Usage, Cost, and Performance? Never fear, you of course could always go straight to the source …

SELECT * FROM system.ai_gateway.usage;

Build out your own or use some default Dashboards.

Wonder who’s accessing what and what those little stinkers are up to?

SELECT * FROM system.access.audit;

I’m not going to get into the details, that would be for another, how-to, type post, which we can do if you ask me. Remember, I’m just trying to introduce you at the 10,000 foot level to what is available to you; lay out the options so you know where to turn and research. Each one of these topics are very complex and require your full attention to understand if they meet your needs. But, it’s good to know they exist.

Access
  - UC privileges on all AI securables; agents governed as UC models.
  - ABAC GRANT policies cover gateway resource types.
  - On-behalf-of-user execution for MCP.
  - Account-wide foundation model allowlisting via system.ai grants
  
  Traffic and cost
  - Smart Routing (Beta): picks model+agent harness per task by complexity, permissions, budget.
  - Budgets with per-user thresholds/hard spend caps; external-provider monitoring.
  - Rate limits at endpoint/user/service principal/group; traffic splitting; fallbacks.
  - OpenAI-compatible unified API.
  - Route Claude Code through Anthropic; Databricks auths, governance still applies.
  
  Service policies (guardrails)
  - SQL-defined policy functions attached to model or MCP services.
  - Built-ins: PII detection/redaction, content safety, prompt injection, data exfiltration.
  - MCP service policies control which tools an agent may invoke.
  - Blocked interactions return HTTP 200 with a block reason, not 4xx.
  
  Observability
  - Unified trace table (Beta): every request and response across all gateway services into one UC table, OpenTelemetry format
  - Usage tracking system tables; cost attribution by service, target model, principal, tag.
  - Payload logging to governed Delta; usage_context request tags for per-project attribution.

The point of all this is simply to broaden our horizons to what we need to think about, and to view this agentic systems as the complex web of details and requirements they actually are. It’s easy for us to just think API+LLM = done. Well … yes and no, and not really. You need to think about serving an LLM as an architectural decision in and of itself, because it is. Everyone’s use case is different, different orgs care about different things at different levels.

For all AI Unity Gateway features, explore the docs.

Pay per token.

I’m going to do my best to simply unwind the tangled web of spiders that makes up the real difference between all the LLM model serving options on Databricks, but I fear I might not be up to the task. Either way, I’m going going to skim the top layer off pay-per-token and provisioned throughput options. This stuff changes all the time, so I encourage you to do your own research and reading. I tend to think about pay-per-token as the starter pack for LLMs.

The pay-per-token models take the form of Databricks Foundational Models APIs. These are out-of-the-box Endpoints that you do not need need to setup or configure yourself. They are just there. They just work. Hit the API, get a response.

The serverless entrypoint to LLMs on Databricks.

• Databricks offers “priority” pay-per-token requests that get to jump the head of the line, if you need low latency.

These pay-per-token Foundational Models (which can also be provisioned throughput) also conform the OpenAI spec, so calling them is standard. For all you SQL hobbits, you can even do it from a Notebook in your favorite language!

Or just even from a little Python script in a Notebook.

The whole point here is you see what you get, it’s easy and straightforward, it’s obvious why Databricks provides this option and tells you that it’s a great entry-point, but might not be the best choice for high-priority and complex use cases. Remember, these models are shared across the entire Workspace, the logging goes into the same system table for everyone using it. Also, things logged to system tables etc, are locked down to admins only.

Metrics — automatic, no setup
  - system.ai_gateway.usage — token counts, latency, requester, request tags, per request routed through the gateway
  - system.serving.endpoint_usage — same idea for legacy serving endpoints
  - system.serving.served_entities — metadata per served foundation model

The point being is that these pay-per-token Foundational Models are a great way to get started and fulfill requirements for simple uses cases that fit these pattern.

Leave a comment

Provisioned Throughput.

On the other end of the spectrum, the most obvious choice for most people will be to pick a Foundational Model via Provisioned Throughput. You have two compute options within this, that will sound familiar to most people. There is On-Demand and Reserved Capacity options, this makes sense to all you AWS folk who are familiar with compute/EC2.

If you are working on a project where you need BETTER ISOLATION for workloads, for whatever reason … this is the option you want to pick.

New boundaries:

  - Your own inference table per endpoint. Pay-per-token is a shared workspace singleton, so its inference table mixes every team's traffic. A PT endpoint is yours — one table per workload, your catalog/schema,
    your retention, its ACLs owned by whoever created the service.
  - Batch payloads get logged. On pay-per-token, batch inference workloads get usage tracking only — no payload logging. PT logs batch too.
  - Table ownership is yours, not account-admin-gated. The system tables still need account + metastore admin, but the inference table you provision is a normal UC table you own and grant on.

Endpoint-level config pay-per-token can't do
  - Traffic splitting across served entities — canary a new model version, A/B two models, shift percentages. Not available on pay-per-token
  - Endpoint ACLs scoped to one workload rather than workspace-wide
  - Serverless usage policies for custom billing-attribution tags

You can also scale to zero with this setup, which is very nice for predictable workloads. Also, if you are working on this sort of project where you require fine grained control, most likely you will want to read up on inference tables. Debuggig, monitoring, optimization, compliance … these become easy problems to solve with model inference.

Wrapping it up.

I feel as if I didn’t even manage to scratch the surface of LLM and model serving on Databricks APIs/Endpoints. It’s a new and fast changing product(s), that are hard to stay on top of, also each bullet point above would be due a series of deep dive articles to understand in depth and best practice usage. Maybe we can do that in the future. Either way, I just wanted to give you all a 10,000 foot view of what to expect when just starting down the road of LLMs on Databricks.

I don’t expect this article to stay relevant for long, but I think in general, it should at least help you logically start to grasp how to think about LLMS on Databricks and which tools to reach for your specific use case(s).

It’s good to start to think about these highly complex agentic systems, and how they can be a simple API call, but often times are required to turn into something much more serious. We have to think about governance ad permission boundaries, cost and monitoring … these aren’t just side-options, they can make or break a system.

We dig into the changing role of the data engineer, the growing overlap between data and AI engineering, and why strong fundamentals may actually become more important as AI makes writing code cheaper and easier.

Pipeline To Insights

Pipeline To Insights is a community-driven blog by passionate Data Engineers who share real-world experiences, technical tutorials, and personal reflections to inspire growth and continuous learning in the evolving world of data and AI.

By Pipeline to Insights

We also talk about AI agents inside modern data platforms, governance and security, AI-generated code and technical debt, controlling AI costs, the importance of human judgment, and why communication and business skills are becoming increasingly valuable for engineers.

• Erfan also shares the story behind Pipeline to Insights, how writing helped accelerate his own learning, and how consistency helped him grow the publication from its first post in October 2024 into a successful data and AI community.

If you’re a data engineer, analytics engineer, AI engineer, or someone trying to break into data, this conversation offers a grounded look at what skills are likely to matter as AI changes how we build software and data systems.

In part uno of this Back to the Basics series, we reviewed recursion in both SQL and Rust. It’s always somewhat rewarding to remind ourselves of the little things that we fell in love with before agents took over, back in the days where StackOverflow was your last hope. Today we continue this series on basics, the idea being not some LeetCode or DSA endgame, but more of a simple reminder of sometimes fun, but often forgotten knowledge that seems more and more likely to slip into oblivion on the age of Lord Claudius.

So, today we talk about threads, and because we are gluttons for punishment, we shall do so in Rust.

If you are looking for some academic and useless pontificating lecture on concurrency at Google scale, then you have come to the wrong place and should move promptly along. Here, we simply take the view of the average engineer and remind ourselves in some shape or form, of applied basics. I’m just here to remind you that some of these core topics are fun in and of themselves, are approachable by the average programmer. We are not looking to master any topic, but really just poke it with a stick, write some code and lodge it in our brain. Who knows when that knowledge will be required and come handy?

So threads.

Thanks to Delta for sponsoring this newsletter! I use Delta Lake daily, and I believe it represents the future of Data Engineering. Content like this would not be possible without their support. Check out their website below.

Concurrency and Threads.

I can still remember back in the good ole’ days when I used to happily write Python scripts with Notepad++, with nary a care in the world. The simple act of solving a problem with my own two fingers however I saw fit, was a transformative time. It was at such a time I ran into something called ProcessPools in Python. Now back in the day this used to be a heated and hot topic for the GIL and Python haters, but apparently such times are coming to an end.

Without getting us caught in the weeds, I still remember the feeling of joy in slow old Python when realizing I could use ProcessPools in a heavy handed way to speed up the processing of data, files, and all sorts of other things that didn’t matter in the end, but stroked my ego to no end. I’ve learned long ago that there is nothing that will better stir up the angry hornets nest of the internet than posting some performance chart. Let me tell you, you will never do it right, and the internet will tell you so.

Concurrency in programming, at least my silly point of view, at the end of the day is really just making your program “run faster.” What other reason is there worthy of our time? Just do things fast, that’s all that matters.

Fair enough, more precisely, we want to do many things at the same time if possible. It may, or may not, end up making the program “faster”, but you get the drift. We usually end up measuring our progress in such terms. One way in which we can run two things at once is something called threads. This seems to be one of those silly topics that in a never ending complicated way, of course, can change depending on hardware and programming language. Again, I don’t care about all that mumbo-jumbo, I just want to have fun. So what are threads as it relates to concurrency?

Let’s start with Rust, and explore threads and maybe some very basic but useful way we might be able to use them to “speed” something up, or do work at the same time. It’s an interesting topic to see how people approach, we have things like async Tokio in Rust as well as pariter crates, both of them very easy ways to accomplish similar tasks in simple manners. Of course we should use these well oiled and used crates, and in other languages well known packages, but it can’t hurt sometimes to see what happens when we do things ourselves. Even if we do blow our proverbial coding foot off.

use std::thread;

fn main() {
    let t1 = thread::spawn(|| {
        println!("Hello World");
    });

    let t2 = thread::spawn(|| {
        println!("Hello World");
    });

    t1.join().unwrap();
    t2.join().unwrap();
}
...
>> Hello World
>> Hello World

What a simple program right? Who cares how fast we print Hello World, but really, how simple is it for a program to spawn a process that runs at the same time as another? In most languages it will be something easy like this. We have our main process running, and we have our two threads. It can be confusing at first, but we have our main “process” that runs, our program, which spins up to threads, each capable of doing something at the same time as the other.

use std::thread;

fn main() {
    let t1 = thread::spawn(|| {
        println!("Hello World from thread 1");
    });

    let t2 = thread::spawn(|| {
        println!("Hello World from thread 2");
    });
    
    t1.join().unwrap();
    t2.join().unwrap();
    println!("Hello from the main process.");
}

This brings in all sorts of complicated topics, depending on your langauge, of how it all actually works, why in the cast of Rust we need to call .join() etc. Again, I’m not writing this to give you a deep dive into concurrency with threads in Rust in particular, but more of a mental model of what threads can be fun in any language, and where concurrent processing could be used to get things done quickly.

Let’s keep things super simple. What if we have a lot of CSV files we want to process in Rust. Let’s use the Divvy Bike Trips dataset as fun way to test our theory of threads in Rust. We have 4 CSV files we need to process.

Of course in a classic program, we would just do this in the main process, one file at a time.

use std::fs::File;
use std::io::{BufRead, BufReader};
use std::time::Instant;

fn main() {
    let start = Instant::now();

    let files = vec![
        "data/202603-divvy-tripdata.csv",
        "data/202604-divvy-tripdata.csv",
        "data/202605-divvy-tripdata.csv",
        "data/202606-divvy-tripdata.csv",
    ];

    for file in files {
        let file_handle = File::open(file).unwrap();
        let reader = BufReader::new(file_handle);

        let count = reader.lines().count() - 1;

        println!("{}: {} records", file, count);
    }

    println!("Total time: {:?}", start.elapsed());
}

All the runtimes are similar.

danielbeach@Daniels-MacBook-Pro-2 thready % cargo run      
   Compiling thready v0.1.0 (/Users/danielbeach/code/thready)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.60s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 451.051708ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.00s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 408.576541ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.00s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 413.700917ms

Ok, but what if we have 10,000 files to work through on some server, maybe we wouldn’t want to do them one at a time, there must be a way to make this faster, how about using threads to process a few at a time. We have modern hardware. Now, it actually isn’t that complicated of code really, remember our simple printing of Hello World example? Let’s just combine this code that reads a CSV with our threading code.

use std::fs::File;
use std::io::{BufRead, BufReader};
use std::thread;
use std::time::Instant;

fn main() {
    let start = Instant::now();

    let files = vec![
        "data/202603-divvy-tripdata.csv",
        "data/202604-divvy-tripdata.csv",
        "data/202605-divvy-tripdata.csv",
        "data/202606-divvy-tripdata.csv",
    ];

    let mut threads = vec![];

    for file in files {
        let t = thread::spawn(move || {
            let file_handle = File::open(file).unwrap();
            let reader = BufReader::new(file_handle);

            let count = reader.lines().count() - 1;

            println!("{}: {} records", file, count);
        });

        threads.push(t);
    }

    for t in threads {
        t.join().unwrap();
    }

    println!("Total time: {:?}", start.elapsed());
}

And, as one would expect, when we use threads, specifically, one thread per file in this case, we get faster runtimes consistently.

danielbeach@Daniels-MacBook-Pro-2 thready % cargo run      
   Compiling thready v0.1.0 (/Users/danielbeach/code/thready)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.60s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 162.972208ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.00s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 168.68625ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.00s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 168.084459ms

I mean not exactly rocket science to implement or be surprised at uh? I mean if we look at the runtimes we reduced the runtime by more than half, with just a little extra code. But, I’m not saying threads are the silver bullet for every problem. Concurrent bugs are probably some of the hardest to debug, also, not every problem is a concurrency problem to be solved with a thread. For example, many modern computing problems have a lot to do with not CPU bound things (like processing a CSV), but maybe I/O problems, say waiting on a network to fulfill a request. Many programming languages provide things like async/await that would be better suited for such situations.

These are important distinctions, but nonetheless, we did learn that with a little extra code and some threading, we more the doubled the speed of a simple program. Again, in whatever language you find yourself writing in, there are probably packages that have been written that can take your current code, and simply use this idea of concurrency under the hood to make it faster. Many of those tools hide the footguns and complexity away from you, which isn’t a bad thing.

For example, the pariter crate I mentioned earlier in Rust, is a great example where concurrency can be applied in a much simpler and straight forward manner, so we get the best of both worlds. Simple code that is easy to understand and read, yet with the power of concurrency baked in.

use pariter::IteratorExt;
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::time::Instant;

fn main() {
    let start = Instant::now();

    let files = vec![
        "data/202603-divvy-tripdata.csv",
        "data/202604-divvy-tripdata.csv",
        "data/202605-divvy-tripdata.csv",
        "data/202606-divvy-tripdata.csv",
    ];

    files
        .into_iter()
        .parallel_map(|file| {
            let file_handle = File::open(file).unwrap();
            let reader = BufReader::new(file_handle);

            let count = reader.lines().count() - 1;

            (file, count)
        })
        .for_each(|(file, count)| println!("{}: {} records", file, count));

    println!("Total time: {:?}", start.elapsed());
}

The results are much the same as our hand rolled threading, but less ugly. I mean if I was picking, I would choose this code every-time because of its simplicity.

danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
   Compiling crossbeam-utils v0.8.23
   Compiling crossbeam-channel v0.5.17
   Compiling pariter v0.6.0
   Compiling thready v0.1.0 (/Users/danielbeach/code/thready)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 1.45s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 166.762334ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.00s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 167.71775ms
danielbeach@Daniels-MacBook-Pro-2 thready % cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.01s
     Running `target/debug/thready`
data/202603-divvy-tripdata.csv: 317037 records
data/202604-divvy-tripdata.csv: 448252 records
data/202605-divvy-tripdata.csv: 653704 records
data/202606-divvy-tripdata.csv: 762550 records
Total time: 171.465708ms

The end of the matter.

This just goes to show you how important it is to at least be aware of some of the basics of how our computers and servers work that we put to the grindstone on a daily basis. At the same time, you don’t need to be a LeetCode savant or a DSA diva to gasp at a very basic level some of these ideas and put them to good use. Its all well and good if something wants to spend a few weeks and months becoming familiar with the inner workings of threads in Rust so they can look down their long noses from their high tower on the rest of us mortals. Have fun up there by yourself.

The rest of us will simply just break out some pariter crate , or whatever it is for you and your language, and cut the runtime in half for some piece of software that reduces compute usage and cost over the long term; driving real world impact and cost reduction without building something at Google scale or with Google complexity. If you want to do those things, then by all means, don’t let me stop you.

What say you, friend?

Leave a comment

In this episode of the Data Engineering Central Podcast, I sit down with to talk about his journey from writing BASIC on an IBM 386 and working with COBOL mainframes to consulting, becoming a chief architect in Big Tech, and eventually losing his job after seven years.

AsianDadEnergy's Newsletter

This is a very public journal of anxiety, existential dread, and way too much tech knowledge. Basically therapy, but with Wi-Fi.

But the layoff wasn’t the end of the story.

We also get into AI and the future of software engineering, why junior engineers may be particularly vulnerable, whether AI is taking away some of the struggle that traditionally made programmers better, and why communication and other human skills are becoming survival skills for engineers—not just skills you need to get promoted.

Finally, we talk about separating your identity from your job, building multiple streams of income, creating content on YouTube and Substack, and why there is a lot more to life than what happens in front of a computer screen.

https://www.youtube.com/@AsianDadEnergy

Big Tech. AI. Layoffs. Financial independence. Content creation. And figuring out what actually matters when the career you’ve spent decades building suddenly disappears.

I think this one will resonate with a lot of people working in tech right now.

Some years ago, feels like a lifetime, I did a short 4-part series on DSA For The Rest Of Us. It was meant to be a fun little excursion to explore some common DSA concepts, except it was written for the average engineer who doesn’t want to work at FAANG/MAANG or send rockets to the moon. We just want to be reasonably good at our jobs and to do our due diligence in applying fundamental concepts to real-life problems. Maybe we don’t want to crack the coding interview.

If you're bored, here are those articles.

• Introduction to Linked Lists.

• Introduction to Binary Search.

• Introduction to QuickSort.

• Introduction to Graphs and Breadth-First Search

Anywho, since we are all being dumbified by our constant use of AI and will continue to do so for the foreseeable future, I thought it might be fun to return to a few data-specific fundamental concepts and, again, try to apply them in a real-world setting, just as a reminder to Lord Claudius that we humans can indeed rage against the machine when required. Who cares if we are relegated to the digital dungeons for being old curmudgeons who simply like to write code, challenge our minds and fingers, and just do a thing ourselves? I, for one, have zero intention of letting my pea-brain turn into mush while the token factories churn out bits and bytes meant to sweep me away.

Recursion. My old friend. Even in SQL.

Many moons ago when my back didn’t hurt and children were not taking all my money; that means when I was young, free, and without a care in the world … I was working in the world of SQL Server, Mainframes, and Microsoft Access. Then one day an interesting problem popped its ugly head up from the well-trodden, 30-year-old-stained corporate carpet underneath those bug-stained fluorescent lights shining on the heads of an endless cubicle farm of angry middle-aged underlings fighting their way towards 2 weeks of vacation a year. It was called Bill-of-Materials.

It’s a well-known and old problem common to manufacturing environments.

Usually, depending on the application, data location, and logic, it requires some sort of recursion to effectively produce a result. Think about it. You, in your hand or pocket, have the One Ring of Power … I mean, a smartphone … it’s made up of many different components. You could imagine each one of those parts had a part number that, if someone asked you for the part numbers that make up the case/shell, it could be some sort of multi-level flow. Part numbers are used in different spots, each with its own list of part numbers that make it up.

Theoretically we don’t know when we start to do some BOM work, when we will hit the end. Each part number could be made up of other part numbers, which could be made up of other partnumbers. I’ve used recursion in this exact situation myself to solve BOM problems in SQL.

What is recursion?

Recursion is an interesting concept/solution to apply to various data and software problems. If you want the fancy and educational version, you probably wouldn’t be reading this Substack. I an’t fancy. Either way, I’m going to give you my take on recursion based on the various ways I’ve used it over the decades, in both SQL and code. Yes, in SQL, strange I know. So recursion. Let’s pull a definition from that once famous but now long forgotten victim of AI website … geeksforgeeks.org.

Think about your favorite scifi movie or TV show, you know how everyone is always afraid of getting stuck in some infinite loop or warp that cannot be escaped. You know the classic scene, you open a door to a room, go through the door on the other side, and end up back at the first door. That’s what recursion in, and it can be dangerous if handled incorrectly.

For example, go ahead and run this in your Python interpreter.

def add(n):
    return n + add(n + 1)

add(1)

What do you notice interesting about the add function above? The fact that it calls itself in the return statement. Strange uh? Strange but useful.

Read more

We dig into Shehab’s journey from writing C++ as a kid to studying computer science at Berkeley, becoming a founder, and eventually building Sail. We talk about why Spark has remained so sticky, what Rust, Apache Arrow, and DataFusion are changing about data infrastructure, and why rebuilding Spark compatibility turned out to be much harder—and more interesting—than expected.

Most importantly, we explore LakeSail’s bigger idea: what if the data pipelines companies already have could become the foundation for their AI pipelines?

A wide-ranging conversation about Spark, Rust, AI, and where data engineering goes from here.

Open wide. Time to take your medicine; your AI medicine, that is. Remember that pink goop your mother used to shovel down your throat; you millennials know what I’m talking about, who could forget that taste of rubber. Blah. Hopefully this experience will be a little better for you, and me.

I’m trying not to be a glutton for punishment; the pace at which new tools, features, products, models, agents, sparkle-glitter-sunshine SaaS thingy-ma-bobs are lobbed over the fence at us is quite overwhelming. I breathe the same air as you, surf the same LinkedIn and Substack feeds as you, get the same FOMO as you when I see someone I respect using a new thingy-ma-bob that I’m not. What am I missing? Can I keep up? I’m giving it my best shot, week in and week out.

Well, now that I’ve moaned, lamented, and hopefully made that striking connection with you over the invisible tunnel of bits and bytes that separates us, again, I must put my poor calloused hands to the plow and throw myself into the deep end. But not just any deep end, mind you, the Agentic Meta Harness deep end. Yikes. I feel like I’m in Blade Runner 2049; I’m not sure if I’m even human anymore.

So. Meta-harness for agents, namely Omnigent.

Thanks to Delta for sponsoring this newsletter! I use Delta Lake daily, and I believe it represents the future of Data Engineering. Content like this would not be possible without their support. Check out their website below.

We need to talk about Coding Agents.

So … what you on? What do you use? What’s your poison? Have you tried this? Yeah, I’m kinda burnt out on it all too. I don’t want to have a Gas Town, I don’t want to be a loop bro. At some point, you just have to pick, like, your IDE of old, and leave the rest to argue their way into oblivion. Me? I like my good ol’ trusty vim, or if I’m feeling spicy, neovim. Now we’ve moved on to coding agents.

Codex, Claude Code, Cursor, the list goes on forever. Not only that, we have to keep up with /skills, MD files, plugins, MCP Servers, and all the rest. It’s a mess, a literal spaghetti of tools, instructions, and skills, all plugged into various and numerous tools. We want different things at different times; we want these repos to be treated one way with a certain set up rules, this other stuff over here we want it done a certain way, maybe we're doing some high-level project planning where we have a flow, could be this stuff over in the corner needs heavy control for running bash commands.

Like I said, a mess.

If we are honest with each other, we are standing on shifting ground; tools are coming and going; we are unsettled and find useful a various assembly of different plugins, skills, MD files, whatever. We have our own little “agentic workflows,” if I may call them that. I prompt this model a certain way; that one differently; I treat one codebase a specific way, and use a different one another based on past technical debt or just its nature.

This is what Omnigent solves.

It’s as easy as …

curl -fsSL https://omnigent.ai/install.sh | sh

Wala, with this simple command, you have installed your first meta-harness, a piece of software that can sit above your various LLM models, workflows, MD files, prompts, … whatever it is you are using. It’s like the One Ring of Power, in the darkenss of your terminal Omnigent will bind them.

What’s the TDLR of a meta-harness like Omnigent?

Of course, I’m only scratching the surface of what a meta-harness like Omnigent can do, but you or I want to get lost in the weeds and get distracted. Rather, let’s just install and play around with the major Omnigent features to get a sense of what we are dealing with, and maybe it will give you ideas on how such a meta-harness can fit into your workflow and development lifecycle. The truth of the matter is that, yes, many tools can do complicated and space-age things within the agentic coding space, but it’s the simple things that provide end users real benefits and features that win the day.

Ok, so what? Show me the money.

Yeah, yeah, yeah, so enough yip-yapping; let’s get into Omnigent, get it set up, piddle and poke around, and see what we can find under the rug. Personally, I am actually excited to get this meta-harness running and see how it can ease my agentic development process. Some of the features that are top of mind for me and seem very useful are …

• single interface for multiple models and harnesses

• rules and policies I can apply to specific use cases

• multi-agent workflow for high-level systems planning, research, and design.

That’s probably just because of the nature of the work I end up doing. Big, high-level project planning that includes extensive systems architecture and design. It’s really important to me that the planning and research phase goes well before a particular direction is chosen. Also, on a more practical note, I have particular codebases that I deal with differently, in very specific ways. Being able to set up policies to deal with this would be very helpful.

Blah. Ok, so you say the install.

>> curl -fsSL https://omnigent.ai/install.sh | sh

So what happens next? Well, we need to set up our existing tooling(s) and harnesses, like Claude or whatever, to Omnigent; this should be the first step.

>> omnigent setup

After running this command you should be prompted to setup and connect any models and harnesses you already are using. In my case I connected both Claude Code and CoPilot up to Omnigent. The setup usually can pickup automtically and detect any tools you are already using. As you can see below, it found my Claude Code setup, sensed my CoPilot setup but I also had to get a token for CoPilot for Omnigent to use. Either way, not rocket science, just connect your existing tooling. I also use OpenCode but did not connect it.

At this point you couse run Omnigent ontop of a single harness, and get all the features we talked about previously. The UI, policies, session sharing, mobile access, etc. For example, I can run … (note: must have tmux installed)

omnigent claude

Of course, as you can see in the bottom of that screen, we get a local host WebUI on port 6767 that has the session we just started in the terminal.

So what, we have meta-harness sitting ontop of our typical Agent Harnesse(s), now what? Well, of course this UI gives us the ability to do things with our conversations as if we were say using Cursor or any other UI. You can watch what’s happening in a session, make a comment on some of the work that’s happening, upload files, edit files directly yourself. It’s kinda like having a IDE for your Agents. We could go on forever exploring the features Omnigent has. One of the more interesting and powerful features is Policies.

We can add our own policies, or enable the many pre-built policies that ships with Omnigent that are centered around Cost and Safety. For example, let’s add a custom policy through just the Web UI Chat interface that we want our Agents to ask for permission before making git commits and pushing changes up to remote repos. Also, we will have it ask for permission before running bash commands.

Apparently these are only session scoped policies, and I need to be more specific. Either way, after asking it to commit and push the git changes it made to the repo (updated the README). It did ask for permissions to run the bash and to run the git commands.

Of course we would probably want to make policies that run over many sessions and the sky is the limit as to what kind of policies and limits we could put in place. There is a massive list here of built-in policies. A very nice feature indeed when you are working across many different code bases, work on teams where budgets and token costs are a thing, where you are worried about Agents doing unintended things to filesystems and production environments. You could easily build a number of policies to watch out for errant aws cli commands that are dangerous, or maybe even an Agent trying to update some terraform file and sync it.

Polly: the omnigent multi-agent super power.

Probably the one feature I am most excited about is Polly. Again, this probably has alot to do with my day to day work of planning technical projects from idea to implementation. This involves lots of research, trial and error, exploring different avenues, reviewing materials, and generally this a lot of back-and-forth between some Agent and myself. Don’t worry, I still spend time just reading raw documentation for myself, there is no replacement for understanding deeply different technologies and features.

“Polly is a multi-agent coding orchestrator. It breaks your task into sub-tasks and delegates each one to a different AI agent, with cross-vendor code review built in. This tutorial shows you how it looks in practice.” - omni docs

I’m excited about this feature because when using some specific Agent to explore a topic, I generally do a ton of different back and forths, exploring every angle I can. Some of that work could defiently be saves when the meta-harness can do that for me. Personally, I’ve found that simply one-shot prompting is the worse approach to solving any problems, most models might do Ok, but depending on the subject they might do quite bad. I would expect this multi-agent Polly routing to impove results a lot.

It’s as easy as …

omnigent polly

This is what I asked Polly.

In the WebUI for Omnigent I can click on Agents and see multiple Agents running, solving this problem. Pretty easy and for many situtations this gives me peace of mind that multi-agents will do a reasonable job out of the box planning or solving the problem I want to explore.

It worked of course, no surpise there.

Again, I think that the multi-agents for planning and researching is a underutililzed and easy to adopt superpower for most devs. You know me, I’m a fan of learning and exploring yourself, but this sort of multi-agent Polly workflow with Omnigent can help with systems design, architecture, and generally large scale planning that is important you want to cover all bases.

The scary part.

Well, maybe scary isn’t that right word I’m looking for, I’m not sure how I feel about this feature. What is sold as the ability to collborate from anywhere … aka your phone, well that can be good and bad. Good from a technology perspective, of course in theory we would want to be able to continue long running sets of tasks and work, collaborate with co-workers on high-stakes agentic work. Sure, I get it.

But, we should be honest about the possible pitfalls of being able to run Agents from your phone. From your bed. The airport. The school line. On the couch. At your mom’s house.

Handle such things with care, you don’t need to be productive 24/7, even if you want your agents to run 24/7, I have serious questions about the mental health of someone running agents from their phone at 8 pm at night. You need to touch grass, exercise, give your head and mind a break. I’ve learned from experience that burnout creeps up quickly and quietly, make good choices that are healthy for you and your body.

Anywho, because I love you all very dearly, I will go ahead and setup this feature for myself, Omnigent recommends using railway. You will need to setup a free account to do this. Just a few button clicks for the free account. Point Railway at the Omnigent GitHub repo, authorize via GitHub, and you’re off to the races while it deploys.

Ya know, part of me is happy that the Railway deployment of omnigent failed on the first try. Does that mean I don’t have to do it?

Didn’t matter what I tried, Omnigent wouldn’t deploy; it always failed on the health checks. Which, honestly, is fine with me, I didn’t want to be suckered into coding from my phone. No good. Ok, maybe I can try a little harder, let’s look at the logs.

RuntimeError: DATABASE_URL is required (env), or set `database_uri:` 
in the server config. Accepted forms: 'postgresql+psycopg://user:pw@host:5432/omnigent' 
(explicit psycopg3), or the 'postgres://' / 'postgresql://' URLs emitted by Railway,
 Render, etc.

Looks like there should be a DATABASE_URL as part of this deploy, the instructions never mentioned it. I tried adding the default localhost Postgres ENV Var and re-deployed; it didn’t work. That’s about as far I would mess with it; as I said, I don’t really want to run Agents from my phone.

Share

Thoughts on meta-harness.

Well, we learned a little something, maybe; I did anyways. I know we’ve barely scratched the surface of all the bells and whistles of Omnigent, but that’s ok. I learned it’s easy to use and install, easy to hook up my other coding harnesses too (Claude and CoPilot), and I do like the option of switching between the WebUI and Terminal depending on what kinda work I’m doing.

The policies were probably my favorite part of Omnigent. When working on sensitive projects that require me to run aws cli commands, or mess around with terraform, policies are another great back-stop to keep tight control where that control is warrented.

Second favorite, and a close one at that, is the Polly feature of Omnigent. It’s hard to state how important this feature can be for those of us who do a lot of research, planning, and systems design/architecture. I spend a lot of time ensuring I’m covering all bases, taking the right direction, and that things are in tip-top shape. Being able to have multiple agents allowing for “cross-vendor code review”, is next-level stuff.

The final piece of Omnigent that has become clearer to me with repeated use is the … cleanliness.

Anyways, this is one tool I’m looking forward to using on a daily basis, and getting to know it better will probably expose a lot of cool and unsung features that will improve my agentic workflows.

Leave a comment

Want more? Checkout this video.

The vibes are cheap; even if the per-token cost increases 10x or 100x, it doesn’t matter, the vibes are cheap; at least they are cheap from the viewpoint of those who hold the power. The power to hire and fire, to set technical direction, and to point a finger and send the peons running.

Read more

I’ve been watching this story play on repeat for some time now; the death of tools like Spark and Kafka has been heralded from the treetops for a decade now, yet here we are … living in a Spark and Kafka world. Someone once told someone else that it is hard to kick against the goad. Indeed it is. I don’t see this as any sort of failure or shortcoming, though; it’s just part of the circle of life, and tech. It’s natural.

Think about the wicked sharp edges that drove legions of SQL Server developers and DBAs insane and to an early grave over the last few decades. Yet, there are still plenty of SSIS packages humming happily away in the quiet corners of the world.

The truth of the matter is, once a tool gets its tentacles latched onto the corporate minds of CTOs, architects, and the like … it’s there to stay for better or worse, for some time to come. Spark is like that. It’s the bread and butter of the data community at large, has been for a while, and will continue to be that … even as Polars, DuckDB, Daft, and the like … bite and nip at the heels. Truth be told, I hold a special little place in my cold and stony heart when it comes to Apache Spark. I grew up in data in a time when I remember ssh’ing into various nodes, installing Spark, the correct AWS S3 JARs, adding IP addresses to configs, updating the memory allocations. Basically terrible things.

• Go search through my old blog; you're sure to find some Spark goodies in there.

Anywho, back to the matter at hand. Lakesail.This one has been on my list for some time, but you know how things go: busy, life, this, that, AI. But they are making enough waves now that I owe it to myself to do what I’ve been doing for a long time, namely, kick the tires and poke with a stick … see what crawls out of the dark corners.

I mean, what is it? Well, the Lakesail homepage will give it to you right away, straight to the old eye, which is appreciated. People forever beating around the bush, making you figure out what the crap an Agentic Wizz-Bang-Popper is. What is Lakesail? Spark, without the JVM tax. You know what sounds even better to my old ringing ears? Spark rewritten in Rust. Like a fly to light me.

I know some of you unbelievers and luddites are sitting there rolling your eyes at the words “rewrite” and “Rust” in the same sentence … but there is a reason people are doing it, my friend; just as well give in to it and get on the ole’ bandwagon. The ride is fine.

Of course, I already mentioned it, but the idea of going after Spark, because it’s “slow” and a “pain in the butt,” is nothing new. Some of the criticisms are true; some are not. It depends on the context. Most people don’t work with “big data,” aka under 10 TB of data or way less, so theoretically they don’t need Spark. They just use it because it’s there and everyone else is using it. Herd mentality.

I’ve made the mistake myself of trying to prove to people that you don’t need a big cluster to work on large data. Be warned: when you start messing with people's golden calf, expect to be summarily burned at the stake; be ready to be a heretic.

One can see the major difference between someone saying DuckDB or Polars can replace Spark, which they can, and someone saying that isn’t the whole story. We don’t live and work on data projects in a vacuum. It takes a serious amount of commitment for any Staff Engineer or CTO to look at a group of drooling and keyboard-happy engineers and say, “Yes, that’s a great idea; go ahead and rewrite our entire stack and legacy data pipeline(s) into DuckDB/Polars/Daft.” That isn’t the real world. In the real world, we have all sorts of deadlines, product launches, tech debt, and general mayhem. Sure, we of course WANT to do those projects, but we don't often get the chance.

Back to Lakesail. This is the impressive part, if it turns out to be true, and believe me, I’m going to try it out, hook, line, and sinker. This is the difference maker. Just swap out the engine, leave the code as is.

“Spark Connect protocol means your existing PySpark, Spark SQL, Delta Lake, and Iceberg code runs unchanged, natively, not as an add-on.”

This is a much easier pill to swallow for data teams. At least for those running the show. It’s a hard sell, even with cost savings, to tell someone to rewrite all the code; a much more palatable message is to just swap compute or configs and leave the code be.

Ok, so all that as an introduction; I don’t really know a better way to understand a tool than by getting to work. So, let’s do that.

Exploring Lakesail concepts, open source, paid platform, etc.

It’s going to be hard to understand all the features of any tool in the short time we have, but I will do my best to give you a working understanding of what Lakesail offers from both open- and closed-source perspectives. We will start with the concepts, check out the open-source options, and finish by setting up an account on the Lakesail Platform to see what that is all about.

Ok, what is Lakesail?

It is

• “… distributed multimodal compute engine …”

• built on Apache Arrow

• built on Apache Datafusion

• Spark users can “… switch to Sail without making code changes.”

• Built with Rust.

This probably makes sense to most of you, I would imagine, besides maybe the word multimodal (you can read more about this on the buoyantdata website). The rest of these are kinda obvious and well known to most of the data community. Apache Arrow is the new go-to in-memory data format; Apache DataFusion has been around for a long time, and many tools are built on it. The idea that Lakesail is built with Rust is a given, but the most interesting part is that we could port existing Apache Spark code to run on Lakesail compute, which is enticing indeed.

How can this be?

The handshake between Lakesail and Spark is Spark Connect.

I have written about Spark Connect before, you can check that link out below. Recently, I’ve also used Spark Connect when building Agentic Chatbots and Agents, etc, as it allows for simple communication between anything non-Spark … and well … Spark.

Let’s jump to the TDLR of Spark Connect, because it’s an important part of the Lakesail conversation. Here is that summary taken from the article I wrote about Spark Connect.

What, pray tell is Apache Spark Connect?

“… a decoupled client-server architecture that allows remote connectivity to Spark clusters using the DataFrame API …”

and

“It is a thin API that can be embedded everywhere: in application servers, IDEs, notebooks, and programming languages … “

So, this is key to understanding how something like Lakesail can truly be a built-in replacement for compute/processing for existing Spark pipelines. It’s just a protocol; Lakesail understands this protocol and can translate a Spark query into a Lakesail query. Easy peasy, not really.

Subscribe now

Playing locally with Lakesail.

What I want to do is play with Lakesail locally, just to see and feel what it’s like; then I would like to set up a Lakesail account, connect it to my AWS account, and maybe run the same setup to see what’s crackin'.

uv init testingLakesail
cd testingLakesail 
uv add pysail pyspark-client==4.2.0

Apparently, we could use the command line with sail to drop into the classic interactive Spark shell, with the Sail server running behind it. But I’m of the opinion that most folk don’t use the Spark command line anymore; most people are in Notebooks. You can thank Databricks for that.

sail spark shell

Either way, let’s write some PySpark code that uses Sail; in theory, it should be boring, because … Sail is just running behind the scenes and taking our normal Spark code and executing it against its own engine built on Rust, Arrow, and Datafusion. This will not be a scientific test at all, but I’m new to this, so give me a break. We will run the same normal PySpark script in a Databricks Notebook on Serverless compute, observe the runtime, then run the same script with Sail. This will all be without a classic Spark cluster, in a sense, on small data, but I just want to see what happens.

If I’m feeling spicy later, I can connect my personal AWS account to a new Lakesail account and run a larger dataset with a classic Spark cluster.

Luckily, kind people like Backblaze provide an open-source hard drive dataset as a queryable Apache Iceberg table on S3 or compressed CSV files.

We will use the raw, compressed CSV files, since they should pose no problem for Spark. I can get them into my personal AWS account in an S3 bucket; then we will have the data against which to run our tests of vanilla Spark and LakeSail Spark.

• 4.54 GB (4,536,210,401 bytes) across 365 objects

• 120,191,685 rows (~120.2M)

Now, I know everyone is going to say this is not fair; we should test on some massive dataset to give vanilla Spark a fighting chance. Also, Serverless; who knows what’s going on behind the curtain? Oz is back there pulling levers. I hear you, but I ain’t got the time, and I’m the one who has to spend the money. Easy for you to say Sunny Jim.

The Spark version running in a Databricks Notebook against Serverless is easy enough.

import datetime as dt
import time

import pyarrow.csv as pacsv
import pyarrow.fs as pafs
from pyspark.sql import functions as F

SRC = "s3://confessions-of-a-data-guy/drivestats/csv/*.csv.gz"
DST = "confessions-of-a-data-guy/drivestats/results/failures_by_model_day_serverless.csv"
REGION = "us-east-1"
START = dt.date.today() - dt.timedelta(days=730)

t0 = time.perf_counter()

failures = (
    spark.read.option("header", True)
    .csv(SRC)
    .select(F.to_date("date").alias("date"), "model", F.col("failure").cast("int").alias("failure"))
    .where((F.col("failure") == 1) & (F.col("date") >= F.lit(START)))
    .groupBy("date", "model")
    .agg(F.count(F.lit(1)).alias("failures"))
    .orderBy(F.col("date").desc(), F.col("failures").desc(), "model")
)

table = failures.toArrow()

fs = pafs.S3FileSystem(
    access_key="xxx",
    secret_key="xxxxxxxx",
    region=REGION,
)
with fs.open_output_stream(DST) as sink:
    pacsv.write_csv(table, sink)

print(f"{table.num_rows} rows -> s3://{DST} in {time.perf_counter() - t0:.1f}s")

display(table)

And the results.

2588 rows -> s3://confessions-of-a-data-guy/drivestats/results/failures_by_model_day_serverless.csv in 35.6s
pyarrow.Table
date: date32[day]
model: string
failures: int64 not null
----
date: [[2026-03-31,2026-03-31,2026-03-31,2026-03-31,2026-03-31,...,2025-04-01,2025-04-01,2025-04-01,2025-04-01,2025-04-01]]
model: [["ST8000NM0055","HGST HUH721212ALN604","TOSHIBA MG10ACA20TE","ST16000NM001G","TOSHIBA MG08ACA16TEY",...,"ST8000DM002","WDC WUH722222ALE6L4","HGST HUH721212ALN604","ST8000NM0055","TOSHIBA MG08ACA16TA"]]
failures: [[5,4,3,2,2,...,2,2,1,1,1]]

Ok, for the Sail swap-out, in the Databricks Notebook we need to do this.

%pip install "pysail==0.7.0"
%pip install "pyspark-client==4.2.0"

Here is the same code, with Sail. You can see the differences are only due to starting a Sail Spark Connect server and then using that. Obviously, in the Production setting, we would have a dedicated Sail Spark Connect Session running and serving from some permanent place. And in theory, we wouldn’t have to change the core code for the Spark pipelines, just what they connect to.

# run these first, in their own cell:  %pip install pysail   then   %restart_python

import datetime as dt
import os
import time

import pyarrow.csv as pacsv
import pyarrow.fs as pafs
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

from pysail.spark import SparkConnectServer

SRC = "s3://confessions-of-a-data-guy/drivestats/csv/*.csv.gz"
DST = "confessions-of-a-data-guy/drivestats/results/failures_by_model_day_sail.csv"
REGION = "us-east-1"
START = dt.date.today() - dt.timedelta(days=730)

# Must be set before the server starts.
os.environ["SAIL_SPARK__SESSION_TIMEOUT_SECS"] = "3600"
os.environ["AWS_ACCESS_KEY_ID"] = "xxxx"
os.environ["AWS_SECRET_ACCESS_KEY"] = "xxxxxxxxxxx"
os.environ["AWS_REGION"] = REGION

t0 = time.perf_counter()

server = SparkConnectServer("127.0.0.1", 0)
server.start(background=True)
host, port = server.listening_address
# The notebook's own `spark` is the Databricks session; this one is Sail.
sail = SparkSession.builder.remote(f"sc://{host}:{port}").create()

failures = (
    sail.read.option("header", True)
    .csv(SRC)
    .select(F.to_date("date").alias("date"), "model", F.col("failure").cast("int").alias("failure"))
    .where((F.col("failure") == 1) & (F.col("date") >= F.lit(START)))
    .groupBy("date", "model")
    .agg(F.count(F.lit(1)).alias("failures"))
    .orderBy(F.col("date").desc(), F.col("failures").desc(), "model")
)

table = failures.toArrow()

fs = pafs.S3FileSystem(
    access_key=os.environ["AWS_ACCESS_KEY_ID"],
    secret_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    region=REGION,
)
with fs.open_output_stream(DST) as sink:
    pacsv.write_csv(table, sink)

print(f"{table.num_rows} rows -> s3://{DST} in {time.perf_counter() - t0:.1f}s")

display(table)

sail.stop()
server.stop()

This is where I started to have problems. I could see from the Logs in the Notebook that the Sail server was starting, but it was churning for minutes running in Serverless compute with no end in sight. My guess is that Databricks Serverless is not the place to test the Sail version of this code.

[2026-08-18T00:19:00Z INFO sail_python::spark::server] Starting the Spark Connect server on 127.0.0.1:37959...
[2026-08-18T00:19:00Z INFO sail_session::session_manager::actor::handler] creating session 9c1eba5a-3765-4572-81cf-63201d9a2d0d

Well, in the end it did finish, just much slower. 197.7s

[2026-08-18T00:19:00Z INFO sail_python::spark::server] Starting the Spark Connect server on 127.0.0.1:37959...
[2026-08-18T00:19:00Z INFO sail_session::session_manager::actor::handler] creating session 9c1eba5a-3765-4572-81cf-63201d9a2d0d
2588 rows -> s3://confessions-of-a-data-guy/drivestats/results/failures_by_model_day_sail.csv in 197.7s
pyarrow.Table
date: date32[day]
model: string
failures: int64 not null
----
date: [[2026-03-31,2026-03-31,2026-03-31,2026-03-31,2026-03-31,...,2025-04-01,2025-04-01,2025-04-01,2025-04-01,2025-04-01]]
model: [["ST8000NM0055","HGST HUH721212ALN604","TOSHIBA MG10ACA20TE","ST16000NM001G","TOSHIBA MG08ACA16TEY",...,"ST8000DM002","WDC WUH722222ALE6L4","HGST HUH721212ALN604","ST8000NM0055","TOSHIBA MG08ACA16TA"]]
failures: [[5,4,3,2,2,...,2,2,1,1,1]][2026-08-18T00:22:17Z INFO sail_session::session_manager::actor::handler] removing session 9c1eba5a-3765-4572-81cf-63201d9a2d0d
[2026-08-18T00:22:17Z INFO sail_python::spark::server] Shutting down the Spark Connect server...
[2026-08-18T00:22:17Z INFO sail_python::spark::server] The Spark Connect server has stopped.
[2026-08-18T00:22:17Z WARN sail_server::actor] aborting 4 task(s) for SessionManagerActor

I mean, both scripts worked; in s3 I can see the results from each process.

Even the results of each file match, which we would expect.

I guess I wasn’t expecting Databricks Serverless to beat Sail on speed, but that wasn’t really the point. We did prove that we can take a vanilla Spark data pipeline, point it at a Sail Spark Connect instance, and run the same exact code without touching it, and get the same exact answer as vanilla Spark, except we ran it through a different engine.

Don’t get me wrong, I could have been doing something totally dumb and backward; you don’t know what you don’t know when playing with new tools.

Out of pure curiosity, I ran the same Sail Spark code from the Notebook locally on my machine to see what would happen. That way I could ensure beyond a doubt that the Spark Sail code was indeed using that as the engine, because who knows what’s happening on Databricks Serverless; it could be playing nice or not. I mean, from the Sail logs, it appears like it went through that way, but I’m not 100% sure.

Well, it took even longer locally. 202.0s

 2588 rows -> s3://confessions-of-a-data-guy/drivestats/results/failures_by_model_day_sail.csv in 202.0s

I don’t really take this as a bad sign; I’m just playing around for the first time. Anyone in their right mind knows that for many datasets, Rust + Datafusion + Apache Arrow is WAY faster than Spark for most data workloads. Maybe until you get to the PB level.

Subscribe now

LakeSail as a Data Platform.

I figured since the above didn’t work out that well, I would poke at the SaaS data platform LakeSail offers. My first impressions were very good; they have a clean website and a straightforward, non-confusing onboarding workflow. You would be amazed at how many people get this wrong. LakeSail clearly put some work into this.

When creating a LakeSail stack in AWS, you can see during the setup what is going on behind the scenes, aka, what technologies they are using to wrap Sail up in way we can use it from their SaaS.

• EKS clusters

• VPC and networking (subnets, security groups, load balancers, VPC endpoints)

• IAM role creation for EKS infrastructure and workload roles

• S3 bucket management for workspace data and job artifacts

• Glue Data Catalog management for default catalog

• KMS key management for EKS secrets encryption

• CloudWatch Logs for EKS control plane logging

• CloudFormation stack discovery (read-only, tagged resources only)

Dang, you know what kinda money I spend on you hobbits??!! This is why, if you are a free subscriber to this Newsletter, give me an upgrade and pay the piper.

Look, you can watch my AWS bill go up.

Look Ma, I got a LakeSail account now.

The UI is very nice and clean, and now that we have connected to my AWS account, let’s go ahead and set up some “Compute” and then I would like to re-run the code we wrote against this more production-like Sail system in one of their Notebooks, and see how smooth the experience is.

When I went about setting up a Cluster to use, naming it, selecting my AWS account, region, etc … it wouldn’t let me go forward without filling in IPv4 CIDR Range for it to use in AWS. This was kinda strange; I've never experienced this before on other compute platforms like Polars Cloud, MotherDuck, Databricks, etc.

I mean, clearly LakeSail has access to my AWS account to set up the previous infra; this probably should have been taken care of during that process. I am no AWS expert, especially in networking, but that leaves me little choice but to scroll through my AWS VPCs and pick some unused IPv4 CIDR range?

While in the middle of doing this, apparently my timing is bad. The site was undergoing some maintenance. So, had to wait for things to get stable and come back again.

Ok, now that we are back in business, I can get to creating this Cluster, of course selecting the Budget option.

As a side note, I wasn’t really planning on reviewing LakeSail’s cloud data platform in depth, but since we are already here, notice what menu options, and therefore features, they provide.

We have Jobs, Notebooks, Compute, Catalogs, and Queries … all of which need no explanation for most of us using modern SaaS data tooling. We would expect such features as the minimum groundwork on which to build robust data platforms.

Ok, after a little wait, it looks like your Cluster is deployed and running. 15 Pods! Holy Mother of Pearl. Better get this test done and everything torn down before I go broke.

Next, I need to make a new Compute Profile (I couldn’t create a Notebook without it). We will use Tiny for size.

After that, we can get to making a Notebook from which to run our previous code.

Next, we wait for the Notebook to come alive, which is a strange feeling. Makes sense when you read the note about waiting for a Marimo pod to emerge.

Waiting for Marimo pod to become ready (namespace: lakesail-compute-zkmjymkcwb)…

Now, finally, we can run our previous Sail code and see what's crackin'. I’m only half interested in the runtime, but I’m more interested in this code's ability to run out of the box without issues. Remember, we are just trying to get a feel for working with Sail itself and LakeSail, the data platform as a whole. I mean, we picked a T3 Small for AWS compute, with a whopping 2GB of RAM. So, the fact that Sail can run this code with that small of compute says something in itself. That’s what you call cost savings.

( I had to change from display(table) to print(table). )

It looks like we got about the same runtime, 177.7s, which is great, considering we used the smallest compute on the market.

Ok, I think it’s time to stop our testing of Sail and LakeSail, if they are one and the same, and just make a few comments about what we’ve seen so far.

Share

The TDLR on LakeSail and the future of Spark.

Wow, what can I say? I literally knew nothing about Sail when I opened my Chrome browser to start writing this … and now I feel like I still know nothing about Sail! Hahaha! I sort of say that tongue-in-cheek, but it’s hard to give the proper time and space to brand-new tools like this in only a day or two. But we played enough to discuss some high-level concepts.

Firstly, it’s so interesting that you and I live in the age of the upstarts, all coming yapping and nipping at the heels of Apache Spark. There is nothing inherently wrong with Apache Spark; it’s been the backbone of data and analytics for a decade at this point. But it shows how far that data community has come, and how much the data landscape has changed. Today's data requirements are not the same as they were 10 years ago.

LakeSail puts those needs at the forefront and solves them. We’ve seen these shifts across the tech landscape at large.

• Moving away from the JVM toward Rust.

• Apache Arrow underpinning everything.

• Crunching more data with less compute.

• Reducing compute costs.

• Reducing complexity.

Like all good humans, data folk want their data faster, cheaper, and right now!

LakeSail, the SaaS platform itself, I thought, was wonderfully nice and easy to use. You would be surprised how many people get this wrong. The onboarding experience, the simple UI, and the ease of use and navigation. Providing a Data Platform experience that developers enjoy and find to be a low barrier to entry is priceless. It means LakeSail is doing things and going places.

Sail, as a technology in and of itself, I don’t have much to comment on, other than that it is brilliant. The idea to support the Spark Connect Protocol was a genius move. It takes all the arguments about migration away. There is no code rewrite; lift-and-shift has always been a problem, and Sail solved it. I mean, it’s so easy to use; it makes sense to at least test workloads on LakeSail and see if money can be saved.

• Remember, this is just my opinion; I have no affiliation with LakeSail. I’m allowed to say something is good … when it is good.

The only question left nagging in my mind is the performance. I’m sure that has more to do with me than Sail. I said it before, and it’s true: for most datasets, Rust, Datafusion, and Arrow are going to beat vanilla Spark. I think maybe that says more about Databricks Serverless running Spark, how good that is, if you want to know my opinion. I can like two tools at once.

Let me know what you think in the comments.

Leave a comment