CSV 数据处理流水线

使用标准命令行工具和 Python 处理表格数据（CSV、TSV、JSON、JSON Lines）。除 Python 3 外，无需任何外部依赖。

适用场景

• 用户提供 CSV/TSV/JSON 文件，并要求对其进行分析、转换或生成报告
• 对表格数据执行连接（join）、过滤（filter）、分组（group）或聚合（aggregate）操作
• 在不同格式间相互转换（如 CSV ↔ JSON 等）
• 去重、排序或清洗杂乱数据
• 生成汇总统计信息或结构化报告
• ETL 工作流：从一种格式中抽取（extract），转换（transform），再加载（load）至另一种格式

使用标准工具快速执行常见操作

查看数据

# 预览前 5 行
head -5 data.csv

# 统计行数（不含表头）
tail -n +2 data.csv | wc -l

# 显示列名（表头）
head -1 data.csv

# 统计第 3 列的唯一值数量
tail -n +2 data.csv | cut -d',' -f3 | sort -u | wc -l

使用 awk 过滤

# 筛选第 3 列大于 100 的行（保留表头）
awk -F',' 'NR==1 || $3 > 100' data.csv > filtered.csv

# 筛选第 2 列匹配指定模式的行（保留表头）
awk -F',' 'NR==1 || $2 ~ /pattern/' data.csv > matched.csv

# 对第 4 列求和
awk -F',' 'NR>1 {sum += $4} END {print sum}' data.csv

排序与去重

# 按第 2 列数值升序排序（保留表头）
head -1 data.csv > sorted.csv && tail -n +2 data.csv | sort -t',' -k2 -n >> sorted.csv

# 基于所有列去重（保留表头）
head -1 data.csv > deduped.csv && tail -n +2 data.csv | sort -u >> deduped.csv

# 基于特定列（第 2 列）去重，仅保留首次出现的行
awk -F',' '!seen[$2]++' data.csv > deduped.csv

Python 操作（适用于复杂转换）

读取与初步检查

import csv, json, sys
from collections import Counter

def read_csv(path, delimiter=','):
    """将 CSV/TSV 读取为字典列表。"""
    with open(path, newline='', encoding='utf-8') as f:
        return list(csv.DictReader(f, delimiter=delimiter))

def write_csv(rows, path, delimiter=','):
    """将字典列表写入 CSV 文件。"""
    if not rows:
        return
    with open(path, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=rows[0].keys(), delimiter=delimiter)
        writer.writeheader()
        writer.writerows(rows)

# 快速统计
data = read_csv('data.csv')
print(f"行数: {len(data)}")
print(f"列名: {list(data[0].keys())}")
for col in data[0]:
    non_empty = sum(1 for r in data if r[col].strip())
    print(f"  {col}: {non_empty}/{len(data)} 非空")

过滤与转换

# 过滤行
filtered = [r for r in data if float(r['amount']) > 100]

# 添加计算列
for r in data:
    r['total'] = str(float(r['price']) * int(r['quantity']))

# 重命名列
renamed = [{('new_name' if k == 'old_name' else k): v for k, v in r.items()} for r in data]

# 类型转换
for r in data:
    r['amount'] = float(r['amount'])
    r['date'] = r['date'].strip()

分组与聚合

from collections import defaultdict

def group_by(rows, key):
    """按某列值对行进行分组。"""
    groups = defaultdict(list)
    for r in rows:
        groups[r[key]].append(r)
    return dict(groups)

def aggregate(rows, group_col, agg_col, func='sum'):
    """按分组聚合某一列。"""
    groups = group_by(rows, group_col)
    results = []
    for name, group in sorted(groups.items()):
        values = [float(r[agg_col]) for r in group if r[agg_col].strip()]
        if func == 'sum':
            agg = sum(values)
        elif func == 'avg':
            agg = sum(values) / len(values) if values else 0
        elif func == 'count':
            agg = len(values)
        elif func == 'min':
            agg = min(values) if values else 0
        elif func == 'max':
            agg = max(values) if values else 0
        results.append({group_col: name, f'{func}_{agg_col}': str(agg), 'count': str(len(group))})
    return results

# 示例：按 category 汇总 revenue
summary = aggregate(data, 'category', 'revenue', 'sum')
write_csv(summary, 'summary.csv')

数据集连接（Join）

def inner_join(left, right, on):
    """基于键列对两个数据集执行内连接（inner join）。"""
    right_index = {}
    for r in right:
        key = r[on]
        if key not in right_index:
            right_index[key] = []
        right_index[key].append(r)

    results = []
    for lr in left:
        key = lr[on]
        if key in right_index:
            for rr in right_index[key]:
                merged = {**lr}
                for k, v in rr.items():
                    if k != on:
                        merged[k] = v
                results.append(merged)
    return results

def left_join(left, right, on):
    """左连接（left join）：保留全部左侧行，右侧缺失字段填充为空字符串。"""
    right_index = {}
    right_cols = set()
    for r in right:
        key = r[on]
        right_cols.update(r.keys())
        if key not in right_index:
            right_index[key] = []
        right_index[key].append(r)
    right_cols.discard(on)

    results = []
    for lr in left:
        key = lr[on]
        if key in right_index:
            for rr in right_index[key]:
                merged = {**lr}
                for k, v in rr.items():
                    if k != on:
                        merged[k] = v
                results.append(merged)
        else:
            merged = {**lr}
            for col in right_cols:
                merged[col] = ''
            results.append(merged)
    return results

# 示例
orders = read_csv('orders.csv')
customers = read_csv('customers.csv')
joined = left_join(orders, customers, on='customer_id')
write_csv(joined, 'orders_with_customers.csv')

去重

def deduplicate(rows, key_cols=None):
    """移除重复行。若指定了 key_cols，则仅依据这些列判断重复。"""
    seen = set()
    unique = []
    for r in rows:
        if key_cols:
            key = tuple(r[c] for c in key_cols)
        else:
            key = tuple(sorted(r.items()))
        if key not in seen:
            seen.add(key)
            unique.append(r)
    return unique

# 仅依据 email 列去重
clean = deduplicate(data, key_cols=['email'])

格式转换

CSV 转 JSON

import json, csv

with open('data.csv', newline='', encoding='utf-8') as f:
    rows = list(csv.DictReader(f))

# 输出为对象数组（JSON Array）
with open('data.json', 'w') as f:
    json.dump(rows, f, indent=2)

# 输出为 JSON Lines 格式（每行一个 JSON 对象，支持流式处理）
with open('data.jsonl', 'w') as f:
    for row in rows:
        f.write(json.dumps(row) + '\n')

JSON 转 CSV

import json, csv

with open('data.json') as f:
    rows = json.load(f)

with open('data.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=rows[0].keys())
    writer.writeheader()
    writer.writerows(rows)

JSON Lines 转 CSV

import json, csv

rows = []
with open('data.jsonl') as f:
    for line in f:
        if line.strip():
            rows.append(json.loads(line))

with open('data.csv', 'w', newline='', encoding='utf-8') as f:
    all_keys = set()
    for r in rows:
        all_keys.update(r.keys())
    writer = csv.DictWriter(f, fieldnames=sorted(all_keys))
    writer.writeheader()
    writer.writerows(rows)

TSV 转 CSV

tr '\t' ',' < data.tsv > data.csv

数据清洗常用模式

修复常见 CSV 问题

def clean_csv(rows):
    """修复常见的 CSV 数据质量问题。"""
    cleaned = []
    for r in rows:
        clean_row = {}
        for k, v in r.items():
            # 清除键与值两端的空白字符
            k = k.strip()
            v = v.strip() if isinstance(v, str) else v
            # 标准化空值表示
            if v in ('', 'N/A', 'n/a', 'NA', 'null', 'NULL', 'None', '-'):
                v = ''
            # 标准化布尔值表示
            if v.lower() in ('true', 'yes', '1', 'y'):
                v = 'true'
            elif v.lower() in ('false', 'no', '0', 'n'):
                v = 'false'
            clean_row[k] = v
        cleaned.append(clean_row)
    return cleaned

验证数据类型

def validate_rows(rows, schema):
    """
    根据 schema 验证各行数据。
    schema：字典，形如 column_name -> 'int'|'float'|'date'|'email'|'str'
    返回 (valid_rows, error_rows)
    """
    import re
    valid, errors = [], []
    for i, r in enumerate(rows):
        errs = []
        for col, dtype in schema.items():
            val = r.get(col, '').strip()
            if not val:
                continue
            if dtype == 'int':
                try:
                    int(val)
                except ValueError:
                    errs.append(f"{col}: '{val}' 不是整数")
            elif dtype == 'float':
                try:
                    float(val)
                except ValueError:
                    errs.append(f"{col}: '{val}' 不是浮点数")
            elif dtype == 'email':
                if not re.match(r'^[^@]+@[^@]+\.[^@]+$', val):
                    errs.append(f"{col}: '{val}' 不是有效邮箱地址")
            elif dtype == 'date':
                if not re.match(r'^\d{4}-\d{2}-\d{2}$', val):
                    errs.append(f"{col}: '{val}' 不符合 YYYY-MM-DD 格式")
        if errs:
            errors.append({'row': i + 2, 'errors': errs, 'data': r})
        else:
            valid.append(r)
    return valid, errors

# 使用示例
valid, bad = validate_rows(data, {'amount': 'float', 'email': 'email', 'date': 'date'})
print(f"有效行数: {len(valid)}, 错误行数: {len(bad)}")
for e in bad[:5]:
    print(f"  第 {e['row']} 行: {e['errors']}")

生成报告

以 Markdown 格式生成汇总报告

def generate_report(data, title, group_col, value_col):
    """生成 Markdown 格式的汇总报告。"""
    lines = [f"# {title}", "", f"**总行数**: {len(data)}", ""]

    # 分组汇总
    groups = group_by(data, group_col)
    lines.append(f"## 按 {group_col} 分组")
    lines.append("")
    lines.append(f"| {group_col} | 计数 | 求和 | 平均值 | 最小值 | 最大值 |")
    lines.append("|---|---|---|---|---|---|")

    for name in sorted(groups):
        vals = [float(r[value_col]) for r in groups[name] if r[value_col].strip()]
        if vals:
            lines.append(f"| {name} | {len(vals)} | {sum(vals):.2f} | {sum(vals)/len(vals):.2f} | {min(vals):.2f} | {max(vals):.2f} |")

    lines.append("")
    lines.append(f"*本报告基于 {len(data)} 行数据生成*")
    return '\n'.join(lines)

report = generate_report(data, "销售汇总", "category", "revenue")
with open('report.md', 'w') as f:
    f.write(report)

大文件处理

对于无法一次性全部载入内存的超大文件：

def stream_process(input_path, output_path, transform_fn, delimiter=','):
    """逐行处理 CSV 文件，避免将整个文件加载进内存。"""
    with open(input_path, newline='', encoding='utf-8') as fin, 
         open(output_path, 'w', newline='', encoding='utf-8') as fout:
        reader = csv.DictReader(fin, delimiter=delimiter)
        writer = None
        for row in reader:
            result = transform_fn(row)
            if result is None:
                continue  # 跳过该行
            if writer is None:
                writer = csv.DictWriter(fout, fieldnames=result.keys(), delimiter=delimiter)
                writer.writeheader()
            writer.writerow(result)

# 示例：以流式方式执行过滤与转换
def process_row(row):
    if float(row.get('amount', 0) or 0) < 10:
        return None  # 跳过金额过小的记录
    row['amount_usd'] = str(float(row['amount']) * 1.0)  # 添加计算字段
    return row

stream_process('big_file.csv', 'output.csv', process_row)

使用提示

• 始终检查文件编码：file -i data.csv；对于带 BOM 的 UTF-8 文件，建议使用 encoding='utf-8-sig'
• 导出含逗号的 Excel 数据时，Python 的 csv 模块会自动处理字段引号
• 处理含国际字符的数据时，使用 json.dumps(ensure_ascii=False)
• 处理竖线分隔（pipe-delimited）文件时，在 csv.reader/csv.writer 中指定 delimiter='|'
• 面对极大规模的聚合运算，可考虑使用 Python 内置的 sqlite3 模块：

  sqlite3 :memory: ".mode csv" ".import data.csv t" "SELECT category, SUM(amount) FROM t GROUP BY category;"