Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Expert data format transformation covering CSV parsing edge cases, encoding detection, JSON/XML/Parquet/Avro conversion, data type inference, schema validation, large file handling with streaming parsers, delimiter detection, and handling of malformed data across all common data interchange formats.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
Each of these library teams gives its bots this skill. Import one and look inside first.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
325 lines · 1,213 words
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.
Expert data format transformation covering CSV parsing edge cases, encoding detection, JSON/XML/Parquet/Avro conversion, data type inference, schema validation, large file handling with streaming parsers, delimiter detection, and handling of malformed data across all common data interchange formats.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the CSV Transformer skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
What it covers
Overview
CSV Parsing: The Hard Parts
Data Type Inference
Format Conversions
Schema Validation
Large File Handling
Format Comparison Decision Matrix
When to Use
Output Format
Example
Edge Cases
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 rowimport 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 ',' # fallbackimport 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)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 schemaimport 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()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)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)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))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}'
))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 countfrom 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 | Schema | Compression | Splittable | Column Pruning | Best For |
|---|---|---|---|---|---|
| CSV | No | No (unless gzipped) | Yes (uncompressed) | No | Interchange, human readability |
| JSON/JSONL | Self-describing | No (unless gzipped) | JSONL: Yes | No | APIs, semi-structured data |
| Parquet | Embedded | Snappy/ZSTD/Gzip | Yes | Yes | Analytics, data lakes |
| Avro | Embedded | Deflate/Snappy | Yes | No | Streaming, schema evolution |
| ORC | Embedded | ZLIB/Snappy/LZO | Yes | Yes | Hive ecosystems |
| Arrow/IPC | Embedded | LZ4/ZSTD | No | Yes | In-memory analytics, IPC |
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.
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
# 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]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.
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
SKILL.md
325 lines · 1,213 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
CSV Transformer bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for