▲ transmission // September 18, 2026

Data Warehouse Sync Pipeline: MySQL to PostgreSQL ETL and Indonesian Address Normalization

How I built an ETL pipeline that syncs 9 MySQL databases into a PostgreSQL warehouse and normalizes messy Indonesian addresses for analytics.

data-engineeringetlpostgresqlmysqlcase-study

Data Warehouse Sync Pipeline: MySQL to PostgreSQL ETL and Indonesian Address Normalization

Internship Project @ Duitin — Sanitized Case Study for Portfolio

⚠️ All credentials, IPs, and row counts are redacted. Addresses are dummy data. Shared with team permission — no production data.

Overview

Built an end-to-end ETL pipeline consolidating 9 production MySQL databases into a single PostgreSQL data warehouse (duitin_warehouse) for BI (Metabase), plus an Indonesian address normalizer that standardizes messy free-text addresses into structured components for reporting.

Role: Data / Backend Intern — designed pipeline, CDC triggers, batch sync, and address NLP logic. Duration: Internship project Stack: Python | MySQL | PostgreSQL | psycopg2 | mysql-connector-python | pandas

Problem

  • Data scattered across 9 MySQL DBs: contributor-performance, digital-product, duitin-b2b, messaging, order-management, performance-management, transaction, user-management, wallet-management
  • Addresses entered by users — inconsistent abbreviations (JKT, JAKSEL, BKS, TGR, JOGJA, JABAR), missing structure, mixed languages — impossible to group by city/province for analytics.
  • No single source of truth for reporting.

Architecture (Sanitized)

flowchart TB
    subgraph MySQL["🔵 MySQL — Production"]
        DBs["9 Source DBs"]
    end
    subgraph PG_STAGE["🟢 PostgreSQL — Staging (9 DBs, 1:1 mirror)"]
        PGS["9 Staging DBs"]
    end
    subgraph WAREHOUSE["🟢 PostgreSQL — duitin_warehouse"]
        WH["9 Schemas + normalized_addresses + address_components"]
    end
    subgraph NORM["⚙️ Address Normalization (DB)"]
        AUTO["automated_normalizer.py\nRegionStandardizer + AddressParser\nConfidence: low/medium/high"]
        ORDER["order_db_normalizer.py\nDual-write PG + MySQL"]
    end
    CSV["csv_address_normalizer.py\nOffline CSV → CSV\n(standalone file tool)"]
    style CSV stroke-dasharray: 5 5
    DBs -->|mysql_to_postgre_sync.py\nbatch 100K rows / 5 min, incremental| PGS
    PGS -->|merge_postgre.py\nschema→schema, 4 workers, COPY, UNLOGGED| WH
    PGS --> AUTO & ORDER
    AUTO --> WH
    ORDER --> WH
    WH --> METABASE["📊 Metabase"]

Flow: MySQL (9 DBs) → PostgreSQL Staging (batch 100K/5min) → duitin_warehouse (parallel merge, 4 workers) → Normalizer → Metabase

Env is via config.env (never committed):

MYSQL_HOST=${MYSQL_HOST}
PG_HOST=${PG_HOST}
MYSQL_SYNC_USER=${MYSQL_SYNC_USER}
PG_SYNC_USER=${PG_SYNC_USER}
POLL_INTERVAL=300

What I Built

  1. MySQL → PostgreSQL Sync (mysql_to_postgre_sync.py)

    • Batch fetch 100K rows/batch, execute_values inserts, ON CONFLICT DO NOTHING
    • Incremental logic + retry, per-table timeout, manual commit
    • Loop every 5 minutes via schedule / time.sleep(300)

  2. CDC with MySQL Triggers (trigger_creator.py)

    • Creates _change_log table + INSERT / UPDATE / DELETE triggers per table for change tracking
    • Enables incremental sync without full scans

  3. Warehouse Merge (merge_postgre.py — ~660 LOC)

    • Merges 9 PG staging DBs → 1 duitin_warehouse (schemas preserve source names)
    • Dynamic DDL from information_schema — preserves types, PK, indexes, FK, check constraints
    • Uses COPY TO STDOUT / COPY FROM STDIN via temp file, UNLOGGED tables, synchronous_commit=OFF, work_mem tuning
    • Parallel with ThreadPoolExecutor (max 4 workers) + sequential fallback, incremental skip (source_count <= target_count), graceful Ctrl+C

  4. Indonesian Address Normalizer

    • RegionStandardizer (170+ mappings): JKT/JAKSEL → Jakarta Selatan, BKS → Bekasi, TGR/TANGSEL → Tangerang Selatan, JOGJA → Yogyakarta, JABAR → Jawa Barat, etc. — regex \b boundaries, longest-first, change tracking (match → full_name)
    • AddressParser: extracts kelurahan, kecamatan, kota/kabupaten (+type Kota/Kabupaten), provinsi, kode_pos (5-digit), RT/RW, negara, confidence scoring (≥3 components → high, ≥2 → medium)
    • automated_normalizer.py polls PostgreSQL + order_db_normalizer.py dual-writes PG+MySQL

Before / After (Dummy Data)

Raw Standardized Components
Jl Merdeka No 10 RT02/RW05, BKS, JABAR 17111 Jl Merdeka No 10 RT 02 RW 05, Kota Bekasi, Jawa Barat 17111 kota: Bekasi (Kota), provinsi: Jawa Barat, kode_pos: 17111, RT: 02 RW: 05, confidence: medium
jaksel, kec kebayoran baru, DKI Jakarta 12120, Indonesia Jakarta Selatan, Kecamatan Kebayoran Baru, DKI Jakarta 12120, Indonesia kecamatan: Kebayoran Baru, kota: Jakarta Selatan, provinsi: DKI Jakarta, kode_pos: 12120, confidence: high
tgr selatan, banten 15310 Tangerang Selatan, Banten 15310 kota: Tangerang Selatan, provinsi: Banten, kode_pos: 15310

Snippet (sanitized):

text, changes = standardizer.standardize_region("BKS, JKT Selatan 12430")
# → "Bekasi, Jakarta Selatan 12430", ["BKS → Bekasi", "JKT Selatan → Jakarta Selatan"]

parsed = parser.parse_address(standardized_text)
# → {kecamatan, kota_kabupaten, provinsi, kode_pos, rt, rw, parsing_confidence}

Impact

  • Single source of truth for analytics — warehouse auto-syncs every 5 min, no manual dumps
  • Address grouping now possible by provinsi / kota / kecamatan / kode_pos for Metabase dashboards
  • Offline CSV normalizer enables bulk cleaning without DB

Challenges Solved

  • Large tables: batch 100K + fetchmany + COPY avoids OOM
  • Schema drift: dynamic type mapping from information_schema
  • Dirty addresses: 100+ abbreviations + regex boundaries + longest-first prevents JKT colliding with Jakarta
  • Performance: 4 workers, UNLOGGED, synchronous_commit OFF — 10x faster initial load

What I Learned

CDC design, batch ETL, PostgreSQL COPY & UNLOGGED, ThreadPoolExecutor, and Indonesian NLP edge cases.

Tech

Python, MySQL, PostgreSQL, pandas, psycopg2-binary, mysql-connector-python, dotenv, threading

Contact: joeyvitogunawan@gmail.com | linkedin | github