ETL Pipeline — Technical Documentation¶

Overview¶

3-step wizard that extracts data from 5 source types, transforms it via natural-language instructions (LLM-generated Polars code), and loads it into a file download or PostgreSQL table. Also supports async GPU-backed ETL jobs and export management.

Architecture¶

Frontend (ETLWizard.jsx)
  ├─ Step 1: ExtractStep.jsx → POST /etl/extract (FormData)
  ├─ Step 2: TransformStep.jsx → POST /etl/transform (JSON)
  └─ Step 3: LoadStep.jsx → POST /etl/load (JSON) or client-side download

Marketplace Backend (etl_routes.py — thin proxy)
  ├─ derive_uuids() → PNU tenant_id + user_id
  ├─ _request() → forward to PNU
  └─ Return PNU response directly

PNU Engine (pipelines.py → pipeline_service.py)
  ├─ etl_extract() → Polars (CSV/JSON/Excel), asyncpg/asyncmy, httpx, bs4, boto3
  ├─ etl_transform() → LLM codegen + Polars sandbox
  ├─ etl_load() → PostgreSQL table or file passthrough
  ├─ etl_create_job() / etl_list_jobs() / etl_get_job() — job metadata in PostgreSQL
  ├─ etl_list_exports() / etl_get_export() / etl_download_export() — MinIO export artifacts
  └─ submit_data_processing_job() — async GPU-backed ETL jobs via K8s

Key Files¶

File	Role
`backend/routes/data_pipeline/etl_routes.py`	Marketplace proxy: 12 endpoints (extract, transform, load, jobs CRUD, exports, async GPU jobs)
`backend/routes/data_pipeline/_common.py`	Shared: `derive_uuids()`, `proxy_error()`
PNU: `src/syntera_engine/api/v1/pipelines.py`	PNU REST endpoints for all pipeline tools
PNU: `src/syntera_engine/services/pipeline_service.py`	ETL business logic (extract, transform, load, jobs, exports)
PNU: `src/syntera_engine/services/code_execution_service.py`	Polars sandbox, LLM codegen, `_modernize_polars_code`
PNU: `src/syntera_engine/services/llm_service.py`	`call_llm()` — routes through `inference_resolver`
PNU: `src/syntera_engine/services/data_pipeline/gpu_job_service.py`	Async GPU-backed job submission, status tracking, result finalization
PNU: `src/syntera_engine/models/pipeline.py`	ORM models: `ETLJob`, `ETLExport`, `DataProcessingJob`
`backend/middleware/auth_middleware.py`	Protects `/api/data-pipeline/etl`
`frontend/src/components/marketplace/tools/data-pipelines/etl/ETLWizard.jsx`	3-step wizard with progress indicator
`frontend/src/components/marketplace/tools/data-pipelines/etl/ExtractStep.jsx`	5 source type selectors + config forms
`frontend/src/components/marketplace/tools/data-pipelines/etl/TransformStep.jsx`	NL instruction + result table preview + code viewer
`frontend/src/components/marketplace/tools/data-pipelines/etl/LoadStep.jsx`	3 destination options (file/PostgreSQL/database-disabled)
`frontend/src/api/dataFetching/dataPipelines.js`	`extractData()`, `transformData()`, `loadData()`, `createETLJob()`, `getETLJobStatus()`, `listETLJobs()`

API Endpoints¶

POST /api/data-pipeline/etl/extract¶

Auth: JWT cookie
Content-Type: multipart/form-data
Form fields: source_type (string), config (JSON string), file (optional UploadFile)
Marketplace: Reads UploadFile, base64-encodes content, sends in JSON body to PNU
PNU source types:
files — CSV/JSON/Excel via Polars, sent as base64 in JSON
database — PostgreSQL/MySQL (asyncpg/asyncmy) or MongoDB (motor)
api — REST GET/POST via httpx
web — HTML table scraping via BeautifulSoup
cloud — S3/Azure blob via boto3/azure-storage-blob
SSRF protection: _validate_url() called for api and web sources — blocks loopback/private/link-local/reserved IPs
Post-extract: PNU calls generate_schema() on first 3 rows → returns data + schema + row count
Response: {"status": "success", "data": [...], "schema": [{name, type}], "row_count": N}

POST /api/data-pipeline/etl/transform¶

Auth: JWT cookie
Request: {"data": [...], "schema": [{name, type}], "instruction": "filter where age > 25", "generated_code": ""}
PNU process: Same as cleaning /perform-operations — generates or validates code, modernizes legacy Polars names, executes in sandbox
Response: {"status": "success", "transformed_data": [...], "generated_code": "..."}

POST /api/data-pipeline/etl/load¶

Auth: JWT cookie
Request: {"data": [...], "destination_type": "file|postgresql", "config": {"table_name": "my_table"}}
PNU destinations:
file — returns data in response (frontend handles download)
postgresql — creates etl_export_* table in PNU database, inserts rows
Response: {"status": "success", "destination": "...", "rows_written": N, "data": [...]} (file) or {"status": "success", "destination": "postgresql:table_name", "rows_written": N} (postgresql)

POST /api/data-pipeline/etl/jobs¶

Creates an ETL job record in PNU PostgreSQL etl_jobs table (configs only, not execution)
Request: {"job_name": "...", "extract_config": {...}, "transform_config": {...}, "load_config": {...}}

GET /api/data-pipeline/etl/jobs/{job_id}¶

Returns job record from PNU

GET /api/data-pipeline/jobs¶

Lists user's jobs (max 50, sorted by created_at desc)
Generic endpoint shared across all data-pipeline job types (ETL, cleaning, labeling, streaming)

GET /api/data-pipeline/etl/exports¶

Lists ETL export artifacts stored in MinIO

GET /api/data-pipeline/etl/exports/{export_id}¶

Gets a specific export artifact metadata

GET /api/data-pipeline/etl/exports/{export_id}/download¶

Returns a presigned MinIO URL for downloading the export (default expiry: 1 hour)

Async GPU-Backed Job Endpoints¶

These endpoints proxy to PNU's data processing job API for GPU-accelerated operations. They are generic and shared across all data-pipeline job types (ETL, cleaning, labeling, streaming):

Marketplace Endpoint	PNU Endpoint	Purpose
`POST /api/data-pipeline/etl/jobs`	`POST /api/v1/pipelines/etl/jobs`	Submit ETL job record (configs only)
`GET /api/data-pipeline/etl/jobs/{id}`	`GET /api/v1/pipelines/etl/jobs/{id}`	Get ETL job record
`GET /api/data-pipeline/jobs`	`GET /api/v1/pipelines/jobs`	List jobs (filter by type, status)
`GET /api/data-pipeline/jobs/{id}/status`	`GET /api/v1/pipelines/jobs/{id}`	Get job status
`GET /api/data-pipeline/jobs/{id}/events`	`GET /api/v1/pipelines/jobs/{id}/events`	Stream real-time status via SSE
`GET /api/data-pipeline/jobs/{id}/results`	`GET /api/v1/pipelines/jobs/{id}/results`	Get presigned URL for results
`POST /api/data-pipeline/jobs/{id}/cancel`	`POST /api/v1/pipelines/jobs/{id}/cancel`	Cancel job, free GPU

See GPU-Backed Data Processing for details on resource allocation, K8s job lifecycle, and container worker.

Note: The GET /api/data-pipeline/jobs/{job_id}/status response now includes prompt_tokens, completion_tokens, and total_tokens fields (populated by PNU when a job completes). These feed into the metering system — see Metering below.

LLM Configuration (PNU)¶

Step	Model	Temperature	Max Tokens
Schema inference (after extract)	`deepseek-v4-flash`	0.7	4000
Code generation (transform)	`deepseek-v4-flash`	0.1	3000

Both route through PNU's inference_resolver → model_deployments DB table. No external env vars.

Security Measures (PNU)¶

SSRF: _validate_url() blocks loopback/private/link-local/reserved IPs for API + web sources
Table name validation: ETL load forces etl_export_ prefix, validates ^[a-zA-Z_][a-zA-Z0-9_]*$
Code sandbox: _FORBIDDEN_PATTERNS regex check + _SAFE_BUILTINS allowlist + _strip_imports + _modernize_polars_code
Engine disposal: SQLAlchemy async engine disposed in finally block after DB extract

Polars 1.x Compatibility¶

The sandbox rewrites legacy method names before execution:

Legacy (LLM output)	Modern (polars 1.x)
`.str.strip()`	`.str.strip_chars()`
`.str.lstrip()`	`.str.strip_chars_start()`
`.str.rstrip()`	`.str.strip_chars_end()`
`pd.` (pandas namespace)	`pl.` (polars namespace)
`pandas.`	`polars.`
`.copy()`	`.clone()`

Frontend Flow¶

Step 1 (Extract): Select source type → fill config → click "Continue to Transform" → extractData(formData) → data + schema stored in wizard state → auto-advances
Step 2 (Transform): Type instruction → click "Generate & Execute" → transformData() → result table preview (first 10 rows) + collapsible code viewer → click "Continue to Load"
Step 3 (Load): Select destination → "Execute Pipeline" → file download (client-side) or PostgreSQL write

Database Configuration (for `database` source type)¶

MongoDB: Uses motor client. Config: db_type, database, collection. Reads up to 1000 docs, strips _id.
PostgreSQL: postgresql+asyncpg://user:pass@host:port/db. Config: db_type, host, port, database, username, password, query. Requires asyncpg package.
MySQL: mysql+asyncmy://user:pass@host:port/db. Same config fields. Requires asyncmy package.

Metering¶

ETL async GPU-backed jobs are metered at completion. Metering is triggered by the SSE proxy endpoint when a terminal status event streams through, or by the REST status endpoint during polling fallback.

Recording point: jobs_routes.stream_job_events() (SSE) or jobs_routes.get_job_status() (polling fallback) — when the PNU job status becomes completed, failed, or cancelled, the marketplace calls record_data_pipeline_usage() with the job's token counts, resource allocation, and duration. The call is idempotent (keyed by resource_id == job_id), so SSE + polling fallback do not create duplicate metering records.

Cost formula:

duration_hours = (completed_at - started_at) / 3600
token_cost     = total_tokens × INFERENCE_TOKEN_RATE
resource_cost  = gpu_count > 0
                   ? gpu_count × duration_hours × DATA_PIPELINE_GPU_HOURLY_RATE
                   : cpu_cores × duration_hours × DATA_PIPELINE_CPU_HOURLY_RATE
total_cost     = token_cost + resource_cost

Failed/cancelled jobs are recorded with cost_usd = 0.0 (accountability without charge).

Token counts come from PNU — the GET /api/data-pipeline/jobs/{job_id}/status response now includes prompt_tokens, completion_tokens, and total_tokens fields. These are populated by _finalize_job() in gpu_job_service.py, which extracts the usage dict from the MinIO result JSON and stores it on the DataProcessingJob row.

MongoDB collection: data_pipeline_usage — see Metering Overview for the full schema and data flow.

Known Limitations¶

ETL synchronous jobs feature is backend-only (no UI for job status/history)
database destination type is disabled in frontend ("Coming soon")
Web scraping only parses first <table> on the page
Cloud storage only supports CSV and JSON (no Excel)
ETL job docs store extract_config with potential credentials in plaintext
SSRF protection uses DNS resolution at validation time; TOCTOU race possible if DNS changes between validation and fetch