Worker

The processing worker turns uploaded files into lake tables with Spark and records progress so users see it in the web app.

How it works

1. worker.py polls the ingest database for files with status uploaded (every 5 seconds).
2. It claims the oldest one (FOR UPDATE SKIP LOCKED), derives the dataset table name vendor__project__folder and runs process_csv.py through spark-submit.
3. The Spark job writes, for the folder’s dataset:
   • bronze: every value as text, plus _file_id, _source_file, _ingested_at, _row_id
   • silver: trimmed, blanks to null, types inferred, exact duplicates in the file removed
   • gold: distinct rows across all files of the folder, rebuilt each time
4. The job prints STAGE <name> lines; the worker turns each into a files update and a file_events row (actor worker), so the UI shows progress live. The job ends with RESULT ... or ERROR ....
5. Archive and Delete. When the API sets a file to archiving or deleting, the worker runs remove_file.py: it deletes the file’s rows from bronze and silver, rebuilds gold from silver, drops the tables if nothing is left (so the next file may bring other columns), expires old Iceberg snapshots so the data files really leave storage, optionally deletes the raw original, and queues views built from that dataset for a rebuild. Removals run before new uploads.

One file at a time, on purpose, because gold is rebuilt per dataset and two rebuilds at once would conflict. On start, files stuck in processing go back to uploaded. Today only csv and tsv are processed; other types become quarantined.

Install

Runs on the host with Spark as the systemd service lake-worker, under /opt/lake-worker, with settings in /etc/lake/worker/worker.env. From the repo on your machine:

bash infra/deploy.sh

then on the server:

bash /opt/lake/scripts/install-all.sh 15

Step 15 needs Spark with S3A (steps 07 and 07.5) and the registry (step 14). Check it:

systemctl status lake-worker
journalctl -u lake-worker -f

Settings

Setting	Default	Meaning
SPARK_SUBMIT	/opt/spark/bin/spark-submit	The launcher
SPARK_MASTER	spark://127.0.0.1:7077	The Spark master
POLL_SECONDS	5	How often it looks for work
JOB_TIMEOUT_SECONDS	1800	A Spark job running longer is stopped and the file marked failed
JOB_CORES	2	Cores for each Spark job
PG_HOST, PG_PORT, INGEST_DB_PASSWORD	from cluster.env	Connection to the ingest database

Test it

Standard library only; no Spark or PostgreSQL needed (a fake store and a fake spark-submit):

cd apps/worker
python3 -m unittest discover -s tests -v

Files

File	What
worker.py	The polling loop and all its SQL
process_csv.py	Spark job: one file to bronze, silver, gold
remove_file.py	Spark job: take one file back out (Archive, Delete)
demo/	Two small CSVs to try an upload
tests/	Tests

Where it fits next

The worker is a deliberately simple stand-in. A scheduler such as Dagster is the option to revisit if the lake needs retries, time schedules or ordered rebuilds of views.