csv-transformer

CSV Transformer

Overview

Data format transformation is deceptively complex. What appears to be a simple task -- reading a CSV and writing Parquet -- actually involves dozens of edge cases around encoding, quoting, escaping, type inference, and schema handling. This skill covers the full landscape of data format operations needed for production data engineering.

CSV Parsing: The Hard Parts

RFC 4180 Compliance and Real-World Deviations

import csv
import io

# RFC 4180 standard CSV
# - Fields separated by commas
# - Records terminated by CRLF
# - Fields MAY be enclosed in double quotes
# - Double quotes within quoted fields are escaped by doubling

# Real-world CSV problems and solutions:

# ... (condensed) ...
        # 'utf-8-sig' strips BOM if present, works normally if absent
        reader = csv.DictReader(f)
        for row in reader:
            yield row

Delimiter Detection

import csv
from collections import Counter

def detect_delimiter(filepath, sample_size=10000):
    """
    Detect CSV delimiter by analyzing character frequency patterns.
    Works for comma, tab, pipe, semicolon, and other single-char delimiters.
    """
    candidates = [',', '\t', '|', ';', ':']

    with open(filepath, 'r', encoding='utf-8-sig') as f:
        # ... (condensed) ...
        dialect = csv.Sniffer().sniff(sample, delimiters=',\t|;:')
        return dialect.delimiter
    except csv.Error:
        return ','  # fallback

Encoding Detection

import chardet
from pathlib import Path

def detect_encoding(filepath, sample_size=100000):
    """
    Detect file encoding using chardet with confidence threshold.
    Falls back to common encodings if confidence is low.
    """
    raw = Path(filepath).read_bytes()[:sample_size]

    result = chardet.detect(raw)
# ... (condensed) ...
    with open(input_path, 'r', encoding=source_encoding, errors='replace') as src:
        with open(output_path, 'w', encoding=target_encoding) as dst:
            for chunk in iter(lambda: src.read(65536), ''):
                dst.write(chunk)

Data Type Inference

import re
from datetime import datetime
from typing import Optional, List, Dict, Any

class TypeInferrer:
    """Infer column types from sample data."""

    DATE_FORMATS = [
        '%Y-%m-%d',
        '%m/%d/%Y',
        '%d/%m/%Y',
        # ... (condensed) ...
        for col, values in columns.items():
            schema[col] = TypeInferrer.infer_type(values)

        return schema

Format Conversions

CSV to Parquet

import pyarrow as pa
import pyarrow.csv as pcsv
import pyarrow.parquet as pq

def csv_to_parquet(csv_path, parquet_path, schema=None, chunk_size=100000):
    """
    Convert CSV to Parquet with optimal settings.
    Handles large files via streaming reads.
    """
    read_options = pcsv.ReadOptions(
        block_size=64 * 1024 * 1024,  # 64 MB read blocks
        # ... (condensed) ...
            writer.write_table(table)
    finally:
        if writer:
            writer.close()

JSON to Parquet

import json
import pyarrow as pa
import pyarrow.parquet as pq

def jsonl_to_parquet(jsonl_path, parquet_path, batch_size=50000):
    """Convert JSON Lines (newline-delimited JSON) to Parquet."""
    batch = []
    writer = None

    with open(jsonl_path, 'r', encoding='utf-8') as f:
        for line in f:
            # ... (condensed) ...
        max_level=3,
    )

    df.to_parquet(parquet_path, engine='pyarrow', compression='snappy', index=False)

XML Parsing

import xml.etree.ElementTree as ET
from lxml import etree
import pandas as pd

def xml_to_dataframe(xml_path, record_tag, namespace=None):
    """
    Parse XML to DataFrame. Uses iterparse for memory efficiency.
    """
    nsmap = namespace or {}
    records = []

    # ... (condensed) ...
                record[col_name] = None
        records.append(record)

    return pd.DataFrame(records)

Avro Operations

import fastavro
from fastavro import writer, reader, parse_schema

# Avro schema definition
avro_schema = {
    "type": "record",
    "name": "Customer",
    "namespace": "com.company.data",
    "fields": [
        {"name": "customer_id", "type": "int"},
        {"name": "name", "type": "string"},
        # ... (condensed) ...
                record[name] = coercer(raw) if raw else None
            records.append(record)

    write_avro(records, avro_path, parse_schema(schema))

Schema Validation

import jsonschema
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class ValidationError:
    row: int
    column: str
    value: str
    expected_type: str
    message: str
# ... (condensed) ...
            if allowed and value not in allowed:
                self.errors.append(ValidationError(
                    row_num, col_name, value, f'enum', f'Not in {allowed}'
                ))

Large File Handling

Streaming Processing

def process_large_csv(filepath, process_fn, chunk_size=50000):
    """Process CSV files of any size using chunked reading."""
    import pandas as pd

    results = []
    encoding = detect_encoding(filepath)
    delimiter = detect_delimiter(filepath)

    for chunk in pd.read_csv(
        filepath,
        encoding=encoding,
        # ... (condensed) ...
        while mm.readline():
            count += 1
        mm.close()
    return count

Parallel Processing

from concurrent.futures import ProcessPoolExecutor
import os

def split_and_process(filepath, process_fn, workers=None):
    """Split large CSV and process chunks in parallel."""
    if workers is None:
        workers = os.cpu_count()

    # Split file into chunks by byte offset
    file_size = os.path.getsize(filepath)
    chunk_size = file_size // workers
# ... (condensed) ...

        results = [f.result() for f in futures]

    return results

Format Comparison Decision Matrix

Default recommendation: Use Parquet with Snappy compression for analytical workloads. Use Avro for streaming and message systems. Use CSV only for human interchange or legacy system requirements.

When to Use

Use this skill when:

• Designing or implementing csv transformer solutions
• Reviewing or improving existing csv transformer approaches
• Making architectural or implementation decisions about csv transformer
• Learning csv transformer patterns and best practices
• Troubleshooting csv transformer-related issues

Do NOT use this skill when:

• The question is about a fundamentally different technology domain
• A more specific sibling skill covers the exact topic needed
• The user needs a complete hands-on tutorial rather than expert guidance

Output Format

# Csv Transformer Analysis

## Context Assessment
[Situation summary and constraints]

## Recommended Approach
[Primary recommendation with rationale]

## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]

## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]

## Next Steps
- [Immediate action item]
- [Follow-up action item]

Example

Input: "Help me implement csv transformer for a medium-scale production application"

Output: A structured analysis covering current state assessment, recommended csv transformer approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.

Edge Cases

• Legacy system integration: When csv transformer must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
• Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
• Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
• Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities