Data Pipeline Orchestration

Python integrations between an ERP and external systems: master data snapshots, change detection, controlled delivery, and traceability for every run.

A sanitized case based on a Python integration platform that replaces flows from a previous middleware. Systems, routes, and tables are described generically.

Outcome: TODOE.g. runs per day or % of deliveries that succeed on retry.

Python
SQL
ETL
SAP HANA
Microsoft Fabric

Architecture

Stages of an integration, from source to run log

SourceERP master dataSAP HANA · SQL
SnapshotCopy of the current state in an intermediate databasePython
ChangesPer-row hash and pending flagPython
ValidationPer-record eligibility rulesPython
DeliveryTarget system API, with retries and bounded concurrencyHTTP
Run logState and outcome of each runSQL · Fabric Warehouse

Dependencies

The snapshot must finish before changes are computed.
Delivery only picks records flagged as pending and eligible.
Database connectors are chosen per environment (QA or production) through configuration, not code.

Controls and traceability

Change detection by source row hash: unchanged content is not resent.
Eligibility checks before delivery, replicating the business rules of the previous flow.
Bounded retries and limited concurrency so the target system is not overloaded.
Selective delivery of specific records to reprocess without rerunning the whole batch.

Technical decisions

Compare the new integration with the previous flow (parity analysis) before replacing it.
Store sync state in the database, not in process memory.
Separate extraction, decision, and delivery so each stage can be tested on its own.

Technical problems and solutions

Problem
A payload field was interpreted differently by the target system, which flagged products with the wrong attribute.
Solution
The rule was confirmed with the target system's vendor and the mapping was adjusted to send the expected value.
Root cause confirmed
Problem
The “pending sync” flag depended on the hash of an intermediate table rather than the actual change date in the ERP.
Solution
The behavior was documented, and filtering by source dates through a join was proposed instead of copying them into every snapshot.
Diagnosis documented; refactor proposed

Code sample

Change detection and delivery with retries (illustrative code)python

def detect_changes(rows, previous_hashes):
    """Mark a row as pending only when its source content changed."""
    pending = []
    for row in rows:
        digest = hashlib.sha256(canonical_json(row).encode()).hexdigest()
        if previous_hashes.get(row["id"]) != digest:
            pending.append({**row, "row_hash": digest, "needs_sync": True})
    return pending


def push_pending(records, client, *, max_workers=4, max_retries=3):
    """Send eligible records with bounded concurrency and retries."""
    eligible = [r for r in records if is_eligible(r)]
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        results = list(pool.map(lambda r: send_with_retry(client, r, max_retries), eligible))
    log_run(results)  # one structured entry per execution
    return results

Case limits

The code is illustrative and sanitized: it is not the original source.
BigQuery is not part of this case; I use it in architecture proposals, not in this integration.
No volumes, schedules, or system names are published.