Privacy-First Data Transformation Pipeline

Leveraging AI to privately and dynamically translate data flows between systems – with no hard-coded field mappings and no sensitive data being fed to public AI platforms.

The Challenge

Enterprise systems rare speak the same language.

It is an integration engineer’s job to transform and structure heterogeneous (or, chaotic, to put it mildly) data sets into structured data usable by the destination system.

Exports from the source system are guaranteed to have column names that differ from the destination schema. Not to mention composite values packed into a single column, and fields whose privacy classification is not obvious from the name alone.

Writing static extract-transform-load (ETL) mappings for every source-to-destination permutation is inefficient and the fragile workflow becomes another point of failure.

This proof of concept pipeline demonstrates that an AI / LLM can substitute for static mapping logic — classifying fields, filtering sensitive data before it reaches the transformation layer, and inferring how source values map to a structured target schema, including splitting a composite field into discrete components. The focus is on a common enterprise integration: an HR information system export consumed by a payroll system API.

Architecture

The pipeline uses two sequential LLM calls, separated by a privacy filter. No row data is sent during classification; no unclassified fields enter the transformation step.

Pass 1 reads only the CSV header row — no data values, only column names — and asks the LLM to classify each as PII, SENSITIVE, or PUBLIC. The classifier is given the destination schema’s field names as context, providing grounding when a source field’s name alone might be ambiguous. An operator reviews the classification table and confirms before any row data is processed.

Pass 2 iterates each row, strips any field not classified PUBLIC, and sends the reduced payload alongside a sample destination record. The model infers field mappings, handles naming differences, normalizes enumerated values, and decomposes composite fields — returning a completed object in the target schema’s shape.

Source and Destination

The source is an HR information system export. Employee names are present but have no corresponding fields in the destination schema. The DepartmentPath column carries a composite hierarchy value. The payroll system’s API expects a normalized record with each field named according to its own schema conventions:

FirstName and LastName are classified PII and never reach Pass 2. The remaining fields are renamed to the destination schema’s conventions: EmployeeId → staffRef, JobCode → roleCode, CostCenter → ledgerCode, Province → taxJurisdiction. The composite DepartmentPath value "Engineering / Platform" is split into discrete division and team fields. The enumerated value "Full-time" is normalized to the destination system’s code "FULL". None of this transformation logic is configured — it is inferred entirely from the destination schema’s field names and example values.

Prompt Engineering

The pipeline is four functions and a main execution region. All LLM calls go through a single invoke_llm() helper that handles streaming Server-Sent Events, retries with back-off, and endpoint authentication — keeping the two passes focused on data logic rather than HTTP plumbing:

# Calls the OpenAI-compatible completions endpoint with streaming enabled.
# Retries up to MAX_RETRIES times on failure. Returns the full response as a string.
# -----------------------------------------------------------------------
def invoke_llm(base_url, api_key, model, system_prompt, user_prompt, label,
               MAX_RETRIES=3, MAX_TOKENS=2000, TIMEOUT=60):

    # rstrip("/") prevents a double-slash in the URL that causes a 404.
    url = base_url.rstrip("/") + "/chat/completions"

    # Authorization header is omitted when api_key is blank (local server with no auth).
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = "Bearer " + api_key

    # stream=True returns Server-Sent Events instead of a single JSON blob.
    # temperature=0 keeps output deterministic -- important for JSON responses.
    body = {
        "model":       model,
        "messages":    [
            {"role": "system", "content": system_prompt},
            {"role": "user",   "content": user_prompt},
        ],
        "temperature": 0,
        "max_tokens":  MAX_TOKENS,
        "stream":      True,
    }

Pass 1 builds a system prompt that names the classification labels and lists the destination schema’s required field names, giving the LLM grounding when a source column name is ambiguous:

system_prompt = (
    "You are a data-privacy classifier. "
    "Classify each column as PII, SENSITIVE, or PUBLIC. "
    "Reply with only a valid JSON object. No markdown."
    + target_hint
)
# target_hint when destination keys are supplied:
# "The destination schema requires: contractType, division, ledgerCode ...
#  If a source field is the only plausible source for a required
#  destination field, classify it as PUBLIC."

Pass 2 sends each row’s PUBLIC-only fields alongside the destination schema sample. The LLM returns a completed record, handling field renaming, value normalization, and composite-field decomposition in a single call:

# translate_row() prompt
user_prompt = (
    "Source record:\n"  + json.dumps(public_payload) + "\n\n"
    "Target schema:\n" + json.dumps(sample_schema)
)

# main loop — one LLM call per row, PUBLIC fields only
for row_number, row in enumerate(csv.DictReader(csv_fh), start=1):
    public_payload = build_public_payload(row, field_labels)
    translated     = translate_row(public_payload, sample_schema,
                                    BASE_URL, API_KEY, MODEL, row_number)
    out_fh.write(json.dumps(translated) + "\n")

AI Endpoint Architecture

Because invoke_llm() targets any OpenAI-compatible endpoint, the two passes can be directed at entirely different infrastructure — and in an enterprise deployment, they should be.

Pass 1 runs against Microsoft Foundry Local. Foundry Local is Microsoft’s on-premises inference runtime: model weights execute directly on the host’s hardware, and inference produces no outbound network traffic. Column names alone can reveal the shape of a dataset in regulated industries; running classification locally eliminates that exposure. Foundry Local exposes the same OpenAI-compatible chat completions API as the cloud service, so the pipeline requires no code changes to switch between environments.

Pass 2 runs against Microsoft Foundry (the managed Azure AI inference platform). By the time a row reaches Pass 2, the privacy filter has already stripped every PII and SENSITIVE field — the architecture enforces that boundary in code, not policy. Microsoft Foundry is the appropriate choice for this step: a managed platform with private networking via Azure Virtual Network integration, Azure RBAC for endpoint access control, built-in content safety layers, compliance coverage across SOC 2, ISO 27001, and HIPAA, and a model catalog that includes GPT-4o and the o-series reasoning models suited to complex schema inference.

Organizations with strict data-residency or air-gap requirements can direct both passes to Foundry Local. Those that want the full model catalog and managed SLA for transformation workloads can split the endpoints. Either way, no code changes are required — only the endpoint URL and credentials differ. The full script is available on GitHub (link below).

Output

Given the source HR export, the pipeline produces a JSONL file in the payroll system’s schema. FirstName and LastName are absent. DepartmentPath is split. EmploymentType is normalized to the destination enumeration. All field names follow the destination system’s conventions:

Takeaways

1. An LLM can substitute for hard-coded ETL mapping logic. Given a source record and a sample destination row, it infers field correspondence, handles naming differences, normalizes enumerated values, and decomposes composite fields.
2. Separating classification from transformation is a sound architectural pattern. Pass 1 never sees row data; Pass 2 never makes privacy decisions. Each LLM call has one clearly bounded job, and the privacy filter between them is enforced in code.
3. Composite field decomposition is inferred, not configured. The LLM splits "Engineering / Platform" into discrete division and team fields by reading the destination schema’s field names, with no parsing rules or regular expressions required.
4. Privacy filtering should be treated as the highest priority, not an afterthought.

GitHub – randy-humblesystems/csv-to-json-pipeline-v1: Privacy-Aware Data Transformation AI Pipeline

Privacy-Aware Data Transformation AI Pipeline. Contribute to randy-humblesystems/csv-to-json-pipeline-v1 development by creating an account on GitHub.