数据格式与读取：CSV、Excel、JSON

做质控模块，第一道坎不是算法，是数据读取。我见过太多人花大把时间调模型，结果连数据都读不对——编码错、列名乱、类型搞混。说实话，这比写逻辑还让人头大。

今天我们就来搞定三种最常用的格式：CSV、Excel 和 JSON。这三种格式，基本覆盖了生物信息、工业检测和数据分析里 90% 的场景。

注意：本章所有代码示例基于 Python 3.8+，用到的库有 csv、json（标准库）、pandas、openpyxl。别装错版本。
批次质控数据读取知识体系 数据读取引擎 CSV Excel JSON 定界符 / 引号 / 编码 大文件分块 引擎选择 数据类型 多sheet 嵌套结构 Schema验证 统一异常处理 DataFrame 输出

4.1 CSV：最基础也最容易翻车

CSV 看起来简单吧？不就是逗号分隔嘛。但我跟你说，恰恰是因为它“简单”，所以坑最多。我在一个基因测序项目里遇到过，某个实验室导出的 CSV，逗号和引号混用，还带着 BOM 头——结果 pandas 读出来第一列全是乱码。

4.1.1 pandas 读 CSV 的标准套路

我个人习惯用 pandas 的 read_csv()，参数多但可控。你想想看，如果只是用 csv 模块一行行解析，碰上大文件效率还行，但做数据集整合还是 pandas 方便。

import pandas as pd

def read_csv_qc(filepath, **kwargs):
    """
    读取 CSV 并做基础质控
    """
    default_params = {
        'sep': ',',
        'encoding': 'utf-8',
        'header': 0,
        'dtype': str,           # 先全读成字符串，避免类型推断翻车
        'keep_default_na': False,
        'low_memory': False
    }
    default_params.update(kwargs)
    
    try:
        df = pd.read_csv(filepath, **default_params)
        print(f'[OK] 读取完成：{filepath}')
        print(f'     行数：{len(df)}，列数：{len(df.columns)}')
        return df
    except Exception as e:
        print(f'[FAIL] 读取失败：{e}')
        return None
经验之谈：设置 dtype=str 是我自己的习惯。为什么呢？因为质控阶段我们更关心“数据有没有被正确读到”，而不是“它是不是数字”。等清洗完再转类型，心里更有底。

4.1.2 避坑指南：常见的 CSV 幺蛾子

  • 编码问题：UTF-8、UTF-8 BOM、GBK、Latin-1 混着来。我建议先试 UTF-8，报错就切 encoding='utf-8-sig' 或 encoding='latin-1'
  • 引号包裹的字段：某些字段里有逗号，比如 "Smith, John" 整体是一个值。用 quotechar='"' 搞定
  • 空行和注释行：有些 CSV 文件开头有 # 打头的注释，用 comment='#' 自动跳过
  • 列数不一致：某一行多了一个逗号，导致列数对不上。这时候 error_bad_lines=False（旧版）或 on_bad_lines='skip'（新版 pandas）可以跳过
# 我曾经遇到过的真实场景：CSV 元数据混合
data_sample = '''# 批次：2024-03-15 质控数据
# 操作员：李工
样本ID,浓度,纯度,备注
QC001,12.3,98.5,"合格"
QC002,9.8,95.2,待复测
QC003,15.1,97.8,合格
'''

# 用参数组合处理这种混合格式
df = pd.read_csv(
    io.StringIO(data_sample),
    comment='#',
    quotechar='"',
    encoding='utf-8'
)

4.2 Excel：功能强但吃内存

Excel 是目前工业界最主流的质控数据交换格式。说实话，它的优势在于非技术人员也能直接看。但劣势也很明显——一个 10 万行的 Excel 文件，能把你内存吃到怀疑人生。

4.2.1 引擎选择：openpyxl vs xlrd

pandas 读 Excel 依赖后端引擎。我强烈建议你直接装 openpyxl，因为它支持 .xlsx，而且对公式、合并单元格的处理更稳定。

引擎 文件格式支持 性能 推荐场景
openpyxl .xlsx 中等，大文件较慢 日常质控、中小文件
calamine .xlsx / .xls / .xlsb 快，Rust 实现 大文件、跨格式场景
xlrd 仅 .xls 较快 遗留系统 .xls 文件 
def read_excel_qc(filepath, sheet_name=0, **kwargs):
    """
    读取 Excel，自动选择引擎
    """
    default_params = {
        'engine': 'openpyxl',
        'header': 0,
        'dtype': str
    }
    default_params.update(kwargs)
    
    # 如果文件超过 50MB，建议用 calamine
    import os
    fsize = os.path.getsize(filepath)
    if fsize > 50 * 1024 * 1024:
        print('[WARN] 文件超过 50MB，推荐使用 calamine 引擎')
        default_params['engine'] = 'calamine'
    
    try:
        df = pd.read_excel(filepath, sheet_name=sheet_name, **default_params)
        print(f'[OK] 读取完成：{filepath} > sheet: {sheet_name}')
        print(f'     行数：{len(df)}，列数：{len(df.columns)}')
        return df
    except Exception as e:
        print(f'[FAIL] 读取失败：{e}')
        return None
小心合并单元格：Excel 中常见的“标题合并”在 pandas 里会变成 NaN。我的做法是：读取前让同事把合并单元格取消，或者用 pd.read_excel(..., header=None) 后自己拼表头。

4.2.2 多 Sheet 读取策略

批次质控经常是一个批次一个 sheet，或者一个批次日期一个 sheet。用 sheet_name=None 可以一次读完所有 sheet，返回一个 dict。

# 读取所有 sheet
all_sheets = pd.read_excel('batch_qc_data.xlsx', sheet_name=None, engine='openpyxl')

for sheet_name, df in all_sheets.items():
    print(f'Sheet: {sheet_name} - 数据行数: {len(df)}')

4.3 JSON：嵌套结构的读写

JSON 现在越来越常见，尤其是从 API 接口获取的质控结果。为什么用 JSON？因为它可以表达复杂的嵌套关系。比如一个样本的完整质控链路：提取→扩增→测序→分析，每个步骤都有子结果。

4.3.1 标准 JSON 读取

最简单的就是 pd.read_json()。但有一个关键点——如果 JSON 是嵌套的，直接读可能会展开成多层索引，索引是用元组表示的。我个人更倾向于先读成 Python 字典，再决定怎么展平。

import json
import pandas as pd

# 一个典型的批次质控 JSON
batch_json = '''
{
  "batch_id": "B20240315",
  "samples": [
    {"id": "QC001", "concentration": 12.3, "purity": 98.5, "status": "PASS"},
    {"id": "QC002", "concentration": 9.8, "purity": 95.2, "status": "REVIEW"}
  ],
  "meta": {
    "operator": "李工",
    "instrument": "NanoDrop-2000"
  }
}
'''

# 方式一：直接 read_json（适用于较规整的 JSON）
df = pd.read_json(batch_json, orient='records')

# 方式二：先 load 再处理（灵活度高）
data = json.loads(batch_json)
df = pd.DataFrame(data['samples'])
print(df)
一个小技巧：如果你发现 read_json() 的结果不对劲，可以先 json.loads() 然后把结果打印出来看。这比猜参数快得多。

4.3.2 处理深层嵌套

我遇到过最夸张的是一个三层嵌套的 JSON——样本信息包着实验数据，实验数据里又包着质控指标。这个场景下 json_normalize() 是救星。

# 嵌套 JSON 示例
nested_json = {
  "batch": "B01",
  "samples": [
    {
      "id": "S001",
      "qc_metrics": {
        "concentration": 15.2,
        "purity_260_280": 1.85,
        "integrity": {"rqn": 9.5, "method": "RNAScreen"}
      }
    }
  ]
}

# 用 json_normalize 展平（pandas 1.3+ 推荐 pd.json_normalize）
from pandas import json_normalize

records = json_normalize(
    nested_json['samples'],
    sep='_',
    max_level=2
)
print(records.columns.tolist())
# 输出：['id', 'qc_metrics.concentration', 'qc_metrics.purity_260_280', 'qc_metrics.integrity.rqn', 'qc_metrics.integrity.method']

4.3.3 Schema 验证与一致性检查

批次质控的 JSON 经常来自不同系统。我曾经踩过一个大坑——上游系统突然改了字段名，把 purity 改成了 purity_260_280，我们完全没发现，结果后续分析全错了。

从那以后，我写了一个简单的 schema 校验函数：

def validate_qc_json(data, required_fields):
    """
    验证 JSON 是否包含必要的质控字段
    required_fields: list of strings, 支持点分隔
    """
    def check_path(obj, path):
        keys = path.split('.')
        current = obj
        for key in keys:
            if isinstance(current, dict) and key in current:
                current = current[key]
            else:
                return False
        return True
    
    missing = [field for field in required_fields if not check_path(data, field)]
    if missing:
        print(f'[WARN] 缺少字段: {missing}')
        return False
    return True

# 使用示例
schema = ['batch_id', 'samples', 'samples.id', 'samples.concentration']
if not validate_qc_json(data, schema):
    raise ValueError('JSON 结构不符合质控 schema，请检查数据来源')

4.4 三种格式的取舍与总结

做质控模块这么多年，我自己的经验是：

  • CSV：适合原始仪器输出、大规模日志。占空间小，但缺少元数据表达能力
  • Excel：适合研发团队内部协作、汇报。非技术人员可以直接看，但大文件要小心
  • JSON：适合系统间接口传输、复杂嵌套的质控链路。可读性不如 Excel，但程序处理方便
最佳实践：写一个统一的数据读取入口函数，内部根据文件后缀自动选择解析方式。这样你的质控流水线可以无缝切换数据源，不用到处改代码。

嗯，说到底，数据读取这步做好了，后面所有的统计、建模、可视化才有意义。别嫌基础，我见过太多“分析结果对不上原始数据”的教训，十有八九都是第一步读数据时埋下的雷。

← 上一章 📖 返回目录 下一章 →