I Built Airflow, Spark, and Iceberg by Hand in MySQL

I built my market-data platform twice.

Version 2 is the one that looks like a modern data stack. PySpark on a two-node Kubernetes cluster I built myself, an Apache Iceberg lakehouse on MinIO behind a Polaris catalog, Airflow kicking off the runs. It computes 11 technical indicators across 13,000+ symbols every hour.

Version 1 is the one that taught me what all of those tools are actually for. It was a MySQL monolith: 35+ tables, 41 stored procedures, 8 scheduled events, and over 325 million rows in the primary indicator tables. Every transform, every indicator, all the scheduling and coordination ran inside the database, because I wrote it there by hand. At the time I didn’t know most of these patterns had names. I just had problems and MySQL.

All the SQL in this post is real, pulled from the schema dump in the repo’s legacy/ directory. Trimmed for readability, not cleaned up.

The data model

Three tiers. Raw* staging tables took API responses as-is. FinancialData held processed OHLCV bars. Six indicator tables (boilerband, macd, sar, directionalmovement, chaikinoscillator, and the indicator columns on FinancialData itself) held derived data keyed back to the bar that produced it.

Here’s the core table:

CREATE TABLE `FinancialData` (
  `FinancialDataID` int NOT NULL AUTO_INCREMENT,
  `StockID` int NOT NULL,
  `StockDate` datetime DEFAULT NULL,
  `Open`  decimal(20,5), `Low`   decimal(20,5),
  `High`  decimal(20,5), `Close` decimal(20,5),
  `Volume` decimal(20,5),
  `EMA` decimal(20,5), `VWAP` decimal(20,5),
  `RateOfChange` decimal(20,5), `OnBalanceVolume` decimal(20,5),
  PRIMARY KEY (`FinancialDataID`,`StockID`),
  UNIQUE KEY `StockID_Date_Idx` (`StockID`,`StockDate`),
  KEY `idx3` (`StockID`,`FinancialDataID`,`StockDate`,`Close`)
) ENGINE=InnoDB AUTO_INCREMENT=320075156
PARTITION BY HASH (`StockID`) PARTITIONS 100;

That AUTO_INCREMENT=320075156 isn’t a typo. It’s the dump’s own receipt for the row count. The indicator tables carry similar values: chaikinoscillator was at 325,821,136 when I froze it.

Two design decisions in that DDL are worth calling out.

Hash partitioning on StockID. Every read and every write in this system is per-symbol. Hash partitioning 100 ways on StockID means any per-symbol query prunes to a single partition, and four workers processing four different symbols are usually touching four disjoint sets of partitions instead of fighting over one B-tree. Six indicator tables partitioned the same way is about 600 physical partitions holding the data.

The composite primary key is not a choice, it’s a tax. MySQL requires every unique key on a partitioned table to include the partition key. So PRIMARY KEY (FinancialDataID, StockID) and UNIQUE KEY (StockID, StockDate) aren’t modeling decisions, they’re the price of partitioning. It also means you cannot enforce a truly global unique constraint that doesn’t involve StockID. Iceberg’s hidden partitioning is solving exactly this: partition layout stops leaking into your logical schema.

Scheduling without a scheduler

v1 had no orchestrator. What it had was scheduled events firing every 2 seconds, each one gated by an advisory lock so runs couldn’t pile up on each other:

CREATE EVENT ProcessRawDataThread0
ON SCHEDULE EVERY 2 SECOND
DO BEGIN
    IF (IS_FREE_LOCK('ProcessRawDataThread0') = 1) THEN
        SELECT GET_LOCK('ProcessRawDataThread0', 0);
        CALL stocks.processRawTableLoop(4, 0);
        SELECT RELEASE_LOCK('ProcessRawDataThread0');
    END IF;
END

GET_LOCK(name, 0) is a non-blocking try-acquire. If the last cycle is still running, this one skips and tries again in 2 seconds. Four of these events, Thread0 through Thread3, were my worker pool. A scheduler with concurrency control, written in SQL, living inside the storage engine.

Spreading the work

Work distribution was a modulo over a queue. Each worker’s cursor only ever saw its slice:

DECLARE allSymbolsCurs CURSOR FOR
    SELECT Symbol
    FROM rawFinancialDataQueue
    WHERE RawFinancialDataQueueID % iTotalThreads = iRemainder
    ORDER BY InsertedTime ASC;

Advisory locks handled event-level exclusivity, but I also needed table-level exclusivity: two workers must never write the same table at once, even with their event locks held. That was a table called ProcessThreadManager, and claiming a table through it is my favorite piece of code in the whole system:

-- try to claim the table
UPDATE ProcessThreadManager
SET `Table` = 'rawFinancialDataQueue', StartDateTime = NOW()
WHERE ProcessThreadManagerID = ieventID;

DO SLEEP(.25);

-- did exactly one claimant survive?
IF ((SELECT COUNT(*) FROM ProcessThreadManager
     WHERE `Table` = 'rawFinancialDataQueue') = 1) THEN
    -- we own it, do the work
ELSE
    -- someone else claimed it in the same window, back off
    UPDATE ProcessThreadManager SET `Table` = NULL
    WHERE ProcessThreadManagerID = ieventID;
END IF;

Write your claim, wait 250ms, then verify you’re the only claimant. If two workers claimed in the same window, both see a count of 2 and both back off. It’s a poor man’s compare-and-swap with built-in race detection, and I invented it because I kept catching two threads inside the same critical section. Years later I’d learn this family of problems is why coordination services exist, and why Airflow just puts task state behind real database transactions in its metadata DB.

Work-stealing, in a stored procedure

The indicator dispatcher is where it gets fun. Instead of processing indicators in a fixed order and blocking when one was busy, the loop walked the set of active indicators and grabbed whichever one was free:

-- allIndicators temp table = the work remaining for this symbol
WHILE ((SELECT COUNT(*) FROM allIndicators) > 0) DO

    -- Indicator: MACD. Is it still pending, and is the table unclaimed?
    CASE WHEN ((SELECT FIND_IN_SET('macd', TableName) FROM allIndicators ...) > 0
        AND (SELECT COUNT(*) FROM ProcessThreadManager WHERE `Table` = 'macd') = 0) THEN

        -- claim it, verify the claim, then insert only missing rows
        INSERT INTO raw_to_macd(StockID, StockDate, FinancialDataID, MACD, MACD_Signal, MACD_hist, UpdateTime)
        SELECT ald.StockID, ald.StockDateTime, ald.FinancialDataID,
               ald.MACD, ald.MACD_Signal, ald.MACD_hist, NOW()
        FROM alldata ald
        LEFT OUTER JOIN macd md ON ald.FinancialDataID = md.FinancialDataID
        WHERE ald.StockID = iStockID AND md.FinancialDataID IS NULL;
    ...

If MACD’s table is claimed by another worker, the loop falls through and tries SAR. If SAR is held, Bollinger. Whatever’s free gets done, and the WHILE keeps spinning until this symbol’s indicator set is empty. That’s a DAG scheduler executing whichever stage is ready, which is what Spark does, except Spark’s version doesn’t need a FIND_IN_SET.

Look at the insert itself too. LEFT OUTER JOIN ... WHERE md.FinancialDataID IS NULL means only rows that don’t already exist get inserted. That’s the “when not matched then insert” half of a MERGE statement, hand-written as an anti-join, and it’s what made every worker idempotent: rerun a symbol and the join finds nothing new to do.

Getting writes right

Nothing was allowed to touch a final table directly. New data for a symbol landed in a temp table, and rows that already existed were deleted before insert via another anti-join against the unique key:

CREATE TEMPORARY TABLE allDataForSymbol
    SELECT *, iStockID AS StockID
    FROM rawfinancialdata WHERE StockSymbol = vStockSymbol;

-- drop anything we already have (unique key: StockID, StockDate)
DELETE ads
FROM allDataForSymbol ads
INNER JOIN FinancialData fd
    ON fd.StockID = ads.StockID AND fd.StockDate = ads.StockDateTime;

And every transformation ran inside a ledger. Open a transaction row, do the work, then either stamp it with the row count and end time, or throw it away:

INSERT INTO raw_data_to_data_transaction(StockID, StartDateTime, CreatedDate)
VALUES (iStockID, NOW(), NOW());
SELECT LAST_INSERT_ID() INTO iTransactionID;

CALL rawFinancialDataToFinancialData(StockSym, ieventID, iRowsInserted);

IF (iRowsInserted > 0) THEN
    UPDATE raw_data_to_data_transaction
    SET RecordQuantity = iRowsInserted, EndDateTime = NOW()
    WHERE Raw_Data_To_DataID = iTransactionID;
ELSE
    DELETE FROM raw_data_to_data_transaction
    WHERE Raw_Data_To_DataID = iTransactionID;
END IF;

Staging, anti-join dedup, idempotent inserts, and a per-transformation ledger. That whole pile is what Iceberg hands you as table-format guarantees: ACID MERGE, snapshot isolation so readers never see a half-written state, and retries that are safe because a failed commit simply never becomes a snapshot.

Every stored procedure also carried the same error contract, so every failure landed in one table with the object name and the full error message:

DECLARE CONTINUE HANDLER FOR SQLEXCEPTION
BEGIN
    GET DIAGNOSTICS CONDITION 1 @errno = MYSQL_ERRNO, @ErrorMsg = MESSAGE_TEXT;
    INSERT INTO errorlog(ObjectName, ErrorNumber, ErrorMessage, CreatedDate)
    VALUES ('processRawTableLoop', @errno, @ErrorMsg, NOW());
END;

The analytical SQL was the good part

The window functions did real work at this scale, and finding Bollinger Band local extrema is my favorite example. Rank every band value, then walk forward in time keeping only the rows that set a new best rank. Those survivors are the local extrema:

-- rank every upper-band value, highest first
INSERT INTO bbLocalMaxes(bbMaxRank, BoilerBandID, FinancialDataID, StockDate, UpperBandMax)
SELECT DENSE_RANK() OVER (ORDER BY UpperBandMax DESC),
       BoilerBandID, FinancialDataID, StockDate, UpperBandMax
FROM bbTimeFrameMaxes
WHERE UpperBandMax IS NOT NULL;

-- walk forward in time; keep a row only if it beats every rank seen so far
SET @iLastBBMaxRank = 100000000;
UPDATE bbLocalMaxes SET bbLastMax =
    CASE WHEN bbMaxRank < @iLastBBMaxRank
         THEN bbMaxRank AND @iLastBBMaxRank := bbMaxRank
         ELSE NULL END
ORDER BY StockDate ASC;

A DENSE_RANK for the global ordering, a user-variable running minimum for the time walk, and then a LEAD() over the survivors to attach each extremum to the next one. New local max or min means a breakout or an accelerating trend, so this little dance is the core of the Bollinger signal. The MACD side did the same style of work with AVG() OVER (PARTITION BY startRange, endRange ORDER BY StockDate) to get slope per trend range, and LAG() built an explicit parent-child lineage table so every bar knew its temporal predecessor.

In v2 this whole category collapses into Spark window specs and lag() over a DataFrame. Same math, but the user-variable tricks and temp-table choreography disappear.

Where it hit the wall

A full cycle, fetch through computed indicators, took about 12 hours. That was WAY too slow for me.

Here’s the part that surprises people: the database wasn’t the bottleneck. The wall was on the Python side. The fetch layer was an asyncio and multiprocessing worker pool of 25 workers pulling from thirteen API endpoints and feeding rows into MySQL, and the database chewed through its compute faster than Python could feed it.

But speeding up Python wouldn’t have saved the architecture, because the ceiling was still real. All the compute ran on one box: four workers sharing one InnoDB buffer pool, one redo log, one write path, cursors walking one row at a time. MySQL kept up with everything I could throw at it. It was never going to keep up with what Spark does to the same workload, because partitioning spread the data, but it couldn’t spread the box.

So I did what anyone does when they don’t know the answer: a bunch of research. That research led straight into the distributed computing world.

v2, and the part nobody tells you

The same workload in v2 runs in about two minutes. Spark spreads the compute across executors, Iceberg makes the writes transactional, Airflow chains ingest into indicators. 12 hours to 2 minutes.

What nobody tells you is that the distributed versions of these problems still bite, just differently. Scaling from one symbol to the full universe, I hit a 155,000-task partition explosion because union sums partitions across DataFrames. Then a driver OOM, because Spark’s metrics listener was tracking all 155k tasks in driver heap. Then Spark 4.0’s ANSI mode throwing on a divide-by-zero that exactly one symbol out of 9,691 was guaranteed to produce. Swapping hand-rolled primitives for professional tools doesn’t delete the failure modes. You just hit them inside Spark instead of inside your own stored procedures.

The mapping

This is the realization the whole rebuild kept hammering home. Almost every tool in v2 replaced something I had already written a poor man’s version of in v1:

Two v1 ideas were good enough that I rebuilt them on purpose. The centralized error log became fail_log, where a failed fetch lands as rows instead of killing the run. The transaction ledger became run_log, keyed so an Airflow retry is idempotent. And I flipped one behavior deliberately: you saw above that v1 deleted empty transactions. v2 keeps zero-success runs, because that row is how a silent failure becomes visible.

Why build the primitive at all

I wouldn’t ship v1’s architecture today. But when I picked Airflow, I knew exactly what its scheduler was doing, because I had written the event loop it replaced. When Spark shuffles data between executors, I know what the modulo-and-cursor version of that looks like and where it falls over. When Iceberg commits a snapshot, I know how many staging tables, anti-joins, and ledger rows that one commit is standing in for.

The difference between my versions and the professional versions is night and day. But I built mine first, and that’s why I understand theirs.

The full schema, all 41 stored procedures, and the ERD are in the legacy/ directory of the repo if you want to dig through the whole thing.