csv-pipeline — Skillopedia

CSV Data Pipeline Process tabular data (CSV, TSV, JSON, JSON Lines) using standard command-line tools and Python. No external dependencies required beyond Python 3. When to Use - User provides a CSV/TSV/JSON file and asks to analyze, transform, or report on it - Joining, filtering, grouping, or aggregating tabular data - Converting between formats (CSV to JSON, JSON to CSV, etc.) - Deduplicating, sorting, or cleaning messy data - Generating summary statistics or reports - ETL workflows: extract from one format, transform, load into another Quick Operations with Standard Tools Inspect Filter w…

, val):\n errs.append(f\"{col}: '{val}' not email\")\n elif dtype == 'date':\n if not re.match(r'^\\d{4}-\\d{2}-\\d{2}', val):\n errs.append(f\"{col}: '{val}' not YYYY-MM-DD\")\n if errs:\n errors.append({'row': i + 2, 'errors': errs, 'data': r})\n else:\n valid.append(r)\n return valid, errors\n\n# Usage\nvalid, bad = validate_rows(data, {'amount': 'float', 'email': 'email', 'date': 'date'})\nprint(f\"Valid: {len(valid)}, Errors: {len(bad)}\")\nfor e in bad[:5]:\n print(f\" Row {e['row']}: {e['errors']}\")\n```\n\n## Generating Reports\n\n### Summary report as Markdown\n\n```python\ndef generate_report(data, title, group_col, value_col):\n \"\"\"Generate a Markdown summary report.\"\"\"\n lines = [f\"# {title}\", f\"\", f\"**Total rows**: {len(data)}\", \"\"]\n\n # Group summary\n groups = group_by(data, group_col)\n lines.append(f\"## By {group_col}\")\n lines.append(\"\")\n lines.append(f\"| {group_col} | Count | Sum | Avg | Min | Max |\")\n lines.append(\"|---|---|---|---|---|---|\")\n\n for name in sorted(groups):\n vals = [float(r[value_col]) for r in groups[name] if r[value_col].strip()]\n if vals:\n lines.append(f\"| {name} | {len(vals)} | {sum(vals):.2f} | {sum(vals)/len(vals):.2f} | {min(vals):.2f} | {max(vals):.2f} |\")\n\n lines.append(\"\")\n lines.append(f\"*Generated from {len(data)} rows*\")\n return '\\n'.join(lines)\n\nreport = generate_report(data, \"Sales Summary\", \"category\", \"revenue\")\nwith open('report.md', 'w') as f:\n f.write(report)\n```\n\n## Large File Handling\n\nFor files too large to load into memory at once:\n\n```python\ndef stream_process(input_path, output_path, transform_fn, delimiter=','):\n \"\"\"Process a CSV row-by-row without loading entire file.\"\"\"\n with open(input_path, newline='', encoding='utf-8') as fin, \\\n open(output_path, 'w', newline='', encoding='utf-8') as fout:\n reader = csv.DictReader(fin, delimiter=delimiter)\n writer = None\n for row in reader:\n result = transform_fn(row)\n if result is None:\n continue # Skip row\n if writer is None:\n writer = csv.DictWriter(fout, fieldnames=result.keys(), delimiter=delimiter)\n writer.writeheader()\n writer.writerow(result)\n\n# Example: filter and transform in streaming fashion\ndef process_row(row):\n if float(row.get('amount', 0) or 0) \u003c 10:\n return None # Skip small amounts\n row['amount_usd'] = str(float(row['amount']) * 1.0) # Add computed field\n return row\n\nstream_process('big_file.csv', 'output.csv', process_row)\n```\n\n## Tips\n\n- Always check encoding: `file -i data.csv` or open with `encoding='utf-8-sig'` for BOM files\n- For Excel exports with commas in values, the CSV module handles quoting automatically\n- Use `json.dumps(ensure_ascii=False)` for international characters\n- Pipe-delimited files: use `delimiter='|'` in csv.reader/writer\n- For very large aggregations, consider `sqlite3` which Python includes:\n ```bash\n sqlite3 :memory: \".mode csv\" \".import data.csv t\" \"SELECT category, SUM(amount) FROM t GROUP BY category;\"\n ```\n---","attachment_filenames":["_meta.json"],"attachments":[{"filename":"_meta.json","content":"{\n \"owner\": \"gitgoodordietrying\",\n \"slug\": \"csv-pipeline\",\n \"displayName\": \"CSV Data Pipeline\",\n \"latest\": {\n \"version\": \"1.0.0\",\n \"publishedAt\": 1770151669885,\n \"commit\": \"https://github.com/clawdbot/skills/commit/894c035377b300d3635af52ffbb8def15cfd0d0a\"\n },\n \"history\": []\n}\n","content_type":"application/json; charset=utf-8","language":"json","size":293,"content_sha256":"8dc4cfb3ae0fd9ae2fea3dec35aabe7fee6ffcabd5c97882b252b83f57460b54"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"CSV Data Pipeline","type":"text"}]},{"type":"paragraph","content":[{"text":"Process tabular data (CSV, TSV, JSON, JSON Lines) using standard command-line tools and Python. No external dependencies required beyond Python 3.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When to Use","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"User provides a CSV/TSV/JSON file and asks to analyze, transform, or report on it","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Joining, filtering, grouping, or aggregating tabular data","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Converting between formats (CSV to JSON, JSON to CSV, etc.)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Deduplicating, sorting, or cleaning messy data","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Generating summary statistics or reports","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ETL workflows: extract from one format, transform, load into another","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Quick Operations with Standard Tools","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Inspect","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Preview first rows\nhead -5 data.csv\n\n# Count rows (excluding header)\ntail -n +2 data.csv | wc -l\n\n# Show column headers\nhead -1 data.csv\n\n# Count unique values in a column (column 3)\ntail -n +2 data.csv | cut -d',' -f3 | sort -u | wc -l","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Filter with ","type":"text"},{"text":"awk","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Filter rows where column 3 > 100\nawk -F',' 'NR==1 || $3 > 100' data.csv > filtered.csv\n\n# Filter rows matching a pattern in column 2\nawk -F',' 'NR==1 || $2 ~ /pattern/' data.csv > matched.csv\n\n# Sum column 4\nawk -F',' 'NR>1 {sum += $4} END {print sum}' data.csv","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Sort and Deduplicate","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Sort by column 2 (numeric)\nhead -1 data.csv > sorted.csv && tail -n +2 data.csv | sort -t',' -k2 -n >> sorted.csv\n\n# Deduplicate by all columns\nhead -1 data.csv > deduped.csv && tail -n +2 data.csv | sort -u >> deduped.csv\n\n# Deduplicate by specific column (keep first occurrence)\nawk -F',' '!seen[$2]++' data.csv > deduped.csv","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Python Operations (for complex transforms)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Read and Inspect","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import csv, json, sys\nfrom collections import Counter\n\ndef read_csv(path, delimiter=','):\n \"\"\"Read CSV/TSV into list of dicts.\"\"\"\n with open(path, newline='', encoding='utf-8') as f:\n return list(csv.DictReader(f, delimiter=delimiter))\n\ndef write_csv(rows, path, delimiter=','):\n \"\"\"Write list of dicts to CSV.\"\"\"\n if not rows:\n return\n with open(path, 'w', newline='', encoding='utf-8') as f:\n writer = csv.DictWriter(f, fieldnames=rows[0].keys(), delimiter=delimiter)\n writer.writeheader()\n writer.writerows(rows)\n\n# Quick stats\ndata = read_csv('data.csv')\nprint(f\"Rows: {len(data)}\")\nprint(f\"Columns: {list(data[0].keys())}\")\nfor col in data[0]:\n non_empty = sum(1 for r in data if r[col].strip())\n print(f\" {col}: {non_empty}/{len(data)} non-empty\")","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Filter and Transform","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Filter rows\nfiltered = [r for r in data if float(r['amount']) > 100]\n\n# Add computed column\nfor r in data:\n r['total'] = str(float(r['price']) * int(r['quantity']))\n\n# Rename columns\nrenamed = [{('new_name' if k == 'old_name' else k): v for k, v in r.items()} for r in data]\n\n# Type conversion\nfor r in data:\n r['amount'] = float(r['amount'])\n r['date'] = r['date'].strip()","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Group and Aggregate","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from collections import defaultdict\n\ndef group_by(rows, key):\n \"\"\"Group rows by a column value.\"\"\"\n groups = defaultdict(list)\n for r in rows:\n groups[r[key]].append(r)\n return dict(groups)\n\ndef aggregate(rows, group_col, agg_col, func='sum'):\n \"\"\"Aggregate a column by groups.\"\"\"\n groups = group_by(rows, group_col)\n results = []\n for name, group in sorted(groups.items()):\n values = [float(r[agg_col]) for r in group if r[agg_col].strip()]\n if func == 'sum':\n agg = sum(values)\n elif func == 'avg':\n agg = sum(values) / len(values) if values else 0\n elif func == 'count':\n agg = len(values)\n elif func == 'min':\n agg = min(values) if values else 0\n elif func == 'max':\n agg = max(values) if values else 0\n results.append({group_col: name, f'{func}_{agg_col}': str(agg), 'count': str(len(group))})\n return results\n\n# Example: sum revenue by category\nsummary = aggregate(data, 'category', 'revenue', 'sum')\nwrite_csv(summary, 'summary.csv')","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Join Datasets","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def inner_join(left, right, on):\n \"\"\"Inner join two datasets on a key column.\"\"\"\n right_index = {}\n for r in right:\n key = r[on]\n if key not in right_index:\n right_index[key] = []\n right_index[key].append(r)\n\n results = []\n for lr in left:\n key = lr[on]\n if key in right_index:\n for rr in right_index[key]:\n merged = {**lr}\n for k, v in rr.items():\n if k != on:\n merged[k] = v\n results.append(merged)\n return results\n\ndef left_join(left, right, on):\n \"\"\"Left join: keep all left rows, fill missing right with empty.\"\"\"\n right_index = {}\n right_cols = set()\n for r in right:\n key = r[on]\n right_cols.update(r.keys())\n if key not in right_index:\n right_index[key] = []\n right_index[key].append(r)\n right_cols.discard(on)\n\n results = []\n for lr in left:\n key = lr[on]\n if key in right_index:\n for rr in right_index[key]:\n merged = {**lr}\n for k, v in rr.items():\n if k != on:\n merged[k] = v\n results.append(merged)\n else:\n merged = {**lr}\n for col in right_cols:\n merged[col] = ''\n results.append(merged)\n return results\n\n# Example\norders = read_csv('orders.csv')\ncustomers = read_csv('customers.csv')\njoined = left_join(orders, customers, on='customer_id')\nwrite_csv(joined, 'orders_with_customers.csv')","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Deduplicate","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def deduplicate(rows, key_cols=None):\n \"\"\"Remove duplicate rows. If key_cols specified, dedupe by those columns only.\"\"\"\n seen = set()\n unique = []\n for r in rows:\n if key_cols:\n key = tuple(r[c] for c in key_cols)\n else:\n key = tuple(sorted(r.items()))\n if key not in seen:\n seen.add(key)\n unique.append(r)\n return unique\n\n# Deduplicate by email column\nclean = deduplicate(data, key_cols=['email'])","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Format Conversion","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"CSV to JSON","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import json, csv\n\nwith open('data.csv', newline='', encoding='utf-8') as f:\n rows = list(csv.DictReader(f))\n\n# Array of objects\nwith open('data.json', 'w') as f:\n json.dump(rows, f, indent=2)\n\n# JSON Lines (one object per line, streamable)\nwith open('data.jsonl', 'w') as f:\n for row in rows:\n f.write(json.dumps(row) + '\\n')","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"JSON to CSV","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import json, csv\n\nwith open('data.json') as f:\n rows = json.load(f)\n\nwith open('data.csv', 'w', newline='', encoding='utf-8') as f:\n writer = csv.DictWriter(f, fieldnames=rows[0].keys())\n writer.writeheader()\n writer.writerows(rows)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"JSON Lines to CSV","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import json, csv\n\nrows = []\nwith open('data.jsonl') as f:\n for line in f:\n if line.strip():\n rows.append(json.loads(line))\n\nwith open('data.csv', 'w', newline='', encoding='utf-8') as f:\n all_keys = set()\n for r in rows:\n all_keys.update(r.keys())\n writer = csv.DictWriter(f, fieldnames=sorted(all_keys))\n writer.writeheader()\n writer.writerows(rows)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"TSV to CSV","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"tr '\\t' ',' \u003c data.tsv > data.csv","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Data Cleaning Patterns","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Fix common CSV issues","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def clean_csv(rows):\n \"\"\"Clean common CSV data quality issues.\"\"\"\n cleaned = []\n for r in rows:\n clean_row = {}\n for k, v in r.items():\n # Strip whitespace from keys and values\n k = k.strip()\n v = v.strip() if isinstance(v, str) else v\n # Normalize empty values\n if v in ('', 'N/A', 'n/a', 'NA', 'null', 'NULL', 'None', '-'):\n v = ''\n # Normalize boolean values\n if v.lower() in ('true', 'yes', '1', 'y'):\n v = 'true'\n elif v.lower() in ('false', 'no', '0', 'n'):\n v = 'false'\n clean_row[k] = v\n cleaned.append(clean_row)\n return cleaned","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Validate data types","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def validate_rows(rows, schema):\n \"\"\"\n Validate rows against a schema.\n schema: dict of column_name -> 'int'|'float'|'date'|'email'|'str'\n Returns (valid_rows, error_rows)\n \"\"\"\n import re\n valid, errors = [], []\n for i, r in enumerate(rows):\n errs = []\n for col, dtype in schema.items():\n val = r.get(col, '').strip()\n if not val:\n continue\n if dtype == 'int':\n try:\n int(val)\n except ValueError:\n errs.append(f\"{col}: '{val}' not int\")\n elif dtype == 'float':\n try:\n float(val)\n except ValueError:\n errs.append(f\"{col}: '{val}' not float\")\n elif dtype == 'email':\n if not re.match(r'^[^@]+@[^@]+\\.[^@]+

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.