Job Data Pipeline: 5 Boards → MySQL → Postgres

A daily Airflow pipeline that pulls job postings from five boards, lands the raw responses in object storage, cleans and upserts them into the Project 1 schema, tracks which jobs are still open, scores new postings for legitimacy, and streams every change on to Postgres. It started as one DAG for one board and grew into five DAGs, three search targets (DBA, IT Support, Manufacturing) and about 11,000 jobs.

Tools

Tool	Used for
Apache Airflow 2.10.4 (custom image airflow-jobs:2.10.4)	Scheduling the five daily DAGs and the scoring catch-up DAG
Python (urllib, boto3, mysql-connector-python, ftfy, python-jobspy)	Extract, transform, load
MinIO (S3-compatible)	Raw landing zone, one folder per board per run date
MySQL 8.4 (project1_jobs)	Writer database, least-privilege pipeline_writer account
MySQL replica + Debezium + Kafka + JDBC sink	Copies of every change on the replica and in Postgres
Laya (local, Unsloth Studio) and Jev (TypeSafe API)	Scam/legitimacy scoring of new postings
Metabase	"Open vs closed jobs" dashboard on the live table

Steps

1. Extract: land the raw response untouched

What it does: one HTTP GET per board, saved byte-for-byte to MinIO under <board>/<run date>/. A transform bug can be fixed by replaying the stored files instead of calling the boards again.

@task
def extract(ds=None):
    with urllib.request.urlopen(f"{ARBEITNOW_URL}?page=1", timeout=30) as resp:
        raw = resp.read()
    key = f"arbeitnow/{ds}/page_1.json"
    _s3_client().put_object(Bucket=MINIO_BUCKET, Key=key, Body=raw)
    return key

Result: every run is replayable. That paid off on 2026-10-04, when 4,481 job descriptions were backfilled from stored pages without re-scraping.

2. Transform and load: clean, then upsert

What it does:

• Repairs every text field with ftfy (a guard against double-encoded UTF-8).
• Gets lookup IDs in one statement.
• Upserts the job, so a job seen again reopens instead of being skipped.

cur.execute(
    "INSERT INTO companies (name) VALUES (%s) "
    "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)",   # returns the id whether new or existing
    (company_name,),
)
company_id = cur.lastrowid

cur.execute(
    "INSERT INTO jobs (title, company_id, location_id, source_id, remote_type, link, "
    " date_posted, job_field, notes, dedup_key) "
    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) "
    "ON DUPLICATE KEY UPDATE status = 'open', last_seen_at = NOW(), closed_at = NULL",
    (title, company_id, location_id, source_id, remote_type, url,
     date_posted, job_field, notes, dedup_key(company_name, title)),
)

dedup_key is a SHA-256 of the normalized company and title, so the same job posted on two boards collides on the unique index instead of creating a second row. Descriptions go into their own job_descriptions table with INSERT IGNORE, so the first description wins and reruns don't generate extra change events.

Result: 9 sources feed one table. 11,135 jobs now, 4,483 open, 9,752 descriptions.

3. Close what disappeared, delete what's been gone 14 days

What it does:

• Any open job from this board not seen in this run is marked closed. The cutoff is the run's own start time taken from MySQL (SELECT NOW()), not Python's clock, to avoid clock skew.
• After 14 days unseen, the job is deleted. Child rows are deleted first and explicitly.

UPDATE jobs SET status = 'closed', closed_at = NOW()
WHERE source_id = (SELECT id FROM sources WHERE name = %s)
  AND status = 'open' AND last_seen_at < %s;          -- %s = this run's start time

DELETE FROM job_descriptions WHERE job_id IN (
  SELECT id FROM jobs WHERE source_id = (SELECT id FROM sources WHERE name = %s)
    AND last_seen_at < NOW() - INTERVAL 14 DAY);
DELETE FROM jobs WHERE source_id = (SELECT id FROM sources WHERE name = %s)
  AND last_seen_at < NOW() - INTERVAL 14 DAY;

Result: status refreshes daily. A job closed by mistake reopens itself the next time it's seen. Retention keeps the table small: a one-time purge on 2026-10-04 removed 7,563 stale jobs, after a full backup dump.

4. Quality gate

What it does: fails the run if the table is empty or any job has a blank title, so a broken source shows up as a red run, not as silently bad data.

if total == 0:
    raise AirflowFailException("Data quality check failed: jobs table is empty")
if bad_titles > 0:
    raise AirflowFailException(f"Data quality check failed: {bad_titles} job(s) with missing title")

5. More sources and search targets

What it does:

• Boards: added Indeed and LinkedIn through python-jobspy, across 8 countries.
• Shared module: job_shared.py holds the three search targets (DBA, IT Support, Manufacturing) and a title classifier that fills jobs.target.
• Countries: guessed from the location text when the board doesn't give one.
• Duplicates: a fuzzy merge joins the same job across boards (same company key, same city, title similarity ≥ 0.95) and keeps the alternate link in job_links.

Result: DBA 922, IT Support 2,733, Manufacturing 2,598 postings classified. Glassdoor, Google Jobs and ZipRecruiter were tested and return nothing from this server, so they're left out.

6. Scam and legitimacy scoring

What it does: before insert, new postings get a first pass from Laya (a local model, free, nothing leaves the network). Anything suspicious goes to Jev (hosted) for a judgement. Rejects land in a job_rejects quarantine table instead of being deleted.

body = {
  "model": "jev-latest",
  "state": f"Title: {title}\nCompany: {company}\n\n{description}",
  "questions": {"legit": {"type": "score",
                          "instructions": "How credible is this as a real job posting?",
                          "criteria": ["obvious scam", "suspicious", "unclear", "credible", "clearly real"]}},
}

Result:

• On a 50-posting test set (40 researched real jobs, 10 known scams), Jev flagged all 10 scams and none of the real jobs (AUC 1.00). Laya scored AUC 0.86.
• Scanning all stored jobs cost about $0.04 and quarantined 1 job.
• Laya was also tested for duplicate detection, scored AUC 0.63, and was dropped. A plain rule did better.

7. Streaming to the replica and Postgres

What it does: MySQL binlog replication keeps a read replica. Debezium reads the same binlog into Kafka, and a JDBC sink writes every insert, update and delete into Postgres (see the CDC project).

Result: writer, replica and Postgres hold the same rows.

Issues encountered

Em dashes turned into â€" on insert

• Cause: mysql.connector without charset="utf8mb4" mangled multi-byte characters. Older rows were already double-encoded at scrape time.
• Fix: set the charset explicitly on the connection, ftfy.fix_text() on every text field, and a one-time repair of 104 old rows.

Every DAG stopped loading for three weeks

• Symptom: No module named 'ftfy'; no new jobs from 2026-09-13 to 2026-10-03.

• Cause: a stack recreate swapped the image back to stock apache/airflow:2.10.4, which doesn't ship ftfy.

• Fix: a two-line custom image, so the dependency can't silently disappear again.

  FROM apache/airflow:2.10.4
  RUN pip install --no-cache-dir ftfy
  

Deleted jobs left orphans in Postgres

• Cause: InnoDB's ON DELETE CASCADE removes child rows without writing them to the binlog, so CDC never sees those deletes. After the purge, Postgres kept 3,879 orphaned descriptions.
• Fix: always delete child rows explicitly before the parent (step 3). Orphans were removed by hand.

DELETE command denied to user 'pipeline_writer'

• Cause: the pipeline account was insert/update only.
• Fix: granted DELETE on exactly the three tables the retention step touches, not on lookups.

Postgres foreign keys halted CDC

• Cause: a 380-company batch arrived in Postgres in a different order across Kafka topics, so child rows hit missing parents and the sink task stopped.
• Fix: dropped the Postgres-side FKs. MySQL stays the source of truth for integrity; definitions are kept in a backup file.

Nightly deadlock at 00:00 UTC

• Cause: two DAGs run at the same minute. close_stale_jobs used an index merge that included a 2-value status index, which locked every source's open rows and deadlocked with another DAG's insert.
• Fix: a composite index (source_id, status, last_seen_at), so the update reads a narrow range (about 121 rows), plus 3 retries on that task.

"Closed" doesn't mean "removed from the board"

• Known limit: most boards are fetched one page per run, so a still-live job pushed off page 1 gets marked closed. Accepted as a proxy and documented. LinkedIn waits 3 days unseen before closing for the same reason.