Data Cleaning Pipeline — Technical Documentation¶

Overview¶

AI-assisted data cleaning tool that infers schema from uploaded JSON, then generates and executes Polars transformation code based on natural-language instructions. All processing happens in the PNU engine; the marketplace is a thin proxy.

Architecture¶

Frontend (DataCleaning.jsx)
  ├─ Stage 1: Upload JSON file (client-side parse)
  ├─ Stage 2: POST /generate_schema → PNU LLM infers column types
  └─ Stage 3: POST /perform_operations → PNU LLM generates Polars code → exec() → result

Marketplace Backend (cleaning_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)
  ├─ clean_generate_schema() → LLM (deepseek-v4-flash) infers column types
  ├─ clean_perform_operations() → LLM generates Polars code → sandbox exec()
  └─ code_execution_service.py — _SAFE_BUILTINS, _FORBIDDEN_PATTERNS, _modernize_polars_code

Key Files¶

File	Role
`backend/routes/data_pipeline/cleaning_routes.py`	Marketplace proxy: 2 endpoints (`/generate_schema`, `/perform_operations`)
`backend/routes/data_pipeline/_common.py`	Shared: `derive_uuids()`, `proxy_error()`
PNU: `src/syntera_engine/api/v1/pipelines.py`	PNU REST endpoints: `/cleaning/generate-schema`, `/cleaning/perform-operations`
PNU: `src/syntera_engine/services/pipeline_service.py`	`clean_generate_schema()`, `clean_perform_operations()`
PNU: `src/syntera_engine/services/code_execution_service.py`	Polars sandbox: `generate_code()`, `execute_code()`, `validate_generated_code()`, `_modernize_polars_code()`, `_SAFE_BUILTINS`, `_FORBIDDEN_PATTERNS`
PNU: `src/syntera_engine/services/llm_service.py`	`call_llm()` — routes through `inference_resolver`
`backend/routes/data_pipeline/__init__.py`	Aggregates routers: `data_pipeline_router` includes cleaning + etl + streaming under `/api/data-pipeline`
`backend/middleware/auth_middleware.py`	Protects `/api/data-pipeline/cleaning`
`frontend/src/components/marketplace/tools/data-pipelines/data-cleaning/DataCleaning.jsx`	3-card UI: upload → schema → transform
`frontend/src/api/dataFetching/dataPipelines.js`	`generateSchema()`, `performOperations()`

API Endpoints¶

POST /api/data-pipeline/cleaning/generate_schema¶

Auth: JWT cookie
Request: {"data": [{"name": "Alice", "age": 30, ...}]}
PNU process: extract_sample_from_json() takes first 3 records → generate_schema() sends to LLM with prompt asking for {schema_list: [{name, type}]} → types normalized via _normalize_type()
Response: {"status": "success", "schema": [{name, type}], "sample_data": [...]}

POST /api/data-pipeline/cleaning/perform_operations¶

Auth: JWT cookie
Request: {"data": [...], "schema": [{name, type}], "input": "filter where age > 25", "generated_code": ""}
PNU process:
Validates + normalizes schema types (str, int, float, bool, datetime)
If generated_code is empty → calls generate_code() (LLM generates Polars code)
If generated_code is provided → calls validate_generated_code() (security check)
_modernize_polars_code() rewrites legacy method names (polars 1.x compat)
execute_code() runs code in sandbox with _SAFE_BUILTINS (no __import__, open, exec, etc.)
Returns transformed data + generated code
Response: {"status": "success", "final_data": [...], "generated_code": "result = pl.DataFrame(data)..."}
Caching: Frontend stores generated_code in state; subsequent transforms reuse it. Changing the instruction clears generatedCode (forces regeneration).

LLM Configuration (PNU)¶

Step	Model	Temperature	Max Tokens
Schema inference	`deepseek-v4-flash`	0.7	4000
Code generation	`deepseek-v4-flash`	0.1	3000

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

Security Measures (PNU `code_execution_service.py`)¶

`validate_generated_code(code)` — called on ALL code paths (user-supplied + LLM-generated)¶

Required patterns: result = and pl.DataFrame( Forbidden patterns: __import__, import os/subprocess/shutil/sys, open(, exec(, eval(, compile(, __builtins__, globals(, locals(, getattr(, setattr(, __class__, __subclasses__, __globals__, __init__, __dict__, pl.read_csv, pl.read_parquet, pl.read_json, pl.read_ipc, pl.read_avro, pl.read_delta, pl.read_database, pl.scan_csv, pl.scan_parquet, pl.scan_ipc, pl.scan_delta, pl.sql

`_strip_imports(code)` — removes all `import` / `from` statements before execution¶

`_modernize_polars_code(code)` — rewrites legacy Polars method names¶

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()`

`execute_code(code, data)` — sandboxed `exec()`¶

env['__builtins__'] = _SAFE_BUILTINS (only: abs, all, any, bool, dict, enumerate, filter, float, int, isinstance, len, list, map, max, min, print, range, round, set, sorted, str, sum, tuple, type, zip)
Only pl (Polars) and data are available as named globals
Code is wrapped in try/except for pl.exceptions.SchemaError and ComputeError
Result must be a Polars DataFrame (auto-converted to list of dicts) or a list of dicts
Fallback: if LLM forgets result variable, uses df as result

Type Normalization¶

_normalize_type() maps LLM-returned types to valid types: string→str, integer→int, boolean→bool, float64→float, timestamp→datetime, etc. Applied both in generate_schema() and perform_operations().

Frontend Flow¶

Card 1 (Data Validator): Upload .json file → parsed client-side → inputData state
Card 2 (Schema Generator): Click "Generate Schema" → calls generateSchema() → schema displayed in <pre>. Checkmark appears when done.
Card 3 (Data Transformer): Type instruction in textarea → Enter to submit → calls performOperations() → result shown + download buttons (JSON/CSV). Changing instruction clears generatedCode.

Known Limitations¶

Only accepts JSON input (no CSV/Excel upload — use ETL for those)
exec() sandbox is regex-based, not a true sandbox — defense in depth, not absolute
The generated_code field is sent to client and could be modified; validate_generated_code mitigates this
Pydantic warning: field name schema in OperationInput shadows BaseModel.schema() (non-breaking)

Metering¶

Async GPU-backed cleaning jobs (submitted via POST /api/data-pipeline/etl/jobs with job_type=cleaning) are metered at completion using the same shared endpoint as ETL and labeling.

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(). The call is idempotent (keyed by resource_id == job_id).

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

Token counts come from PNU — the GET /api/data-pipeline/jobs/{job_id}/status response 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 (accumulated across all LLM calls within the job) and stores it on the DataProcessingJob row.

Note: Synchronous cleaning operations (/generate_schema and /perform_operations called directly, not via async jobs) are not metered through the data pipeline metering system. If they go through the marketplace inference path, token usage is tracked by track_inference_usage() instead.

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