📚 课程目录第4章数据源接入

数据源接入：读取 CSV、Excel、JSON、数据库表

批本质控的第一步，说白了就是数据读取。你流程写得再漂亮，要是数据进不来，一切都白搭。我见过不少初学的朋友，花大量时间研究算法，结果卡在文件编码或数据库连接上。所以这章咱们就专注解决一个问题：如何稳健地把各种数据源拉进 Python 里。

CVS 读取——最常用，也最容易踩坑

CSV 是质控数据最通用的格式。pandas 的 read_csv 很强大，但参数多得一塌糊涂。我自己的习惯是：先不看文档，直接试读，出了错再对症下药。

给你一个我项目里常用的模板：

import pandas as pd

df = pd.read_csv(
    "batch_20250301.csv",
    sep=",",          # 分隔符，有时是制表符 \t
    encoding="utf-8", # Excel 导出的中文经常是 gbk
    dtype={"batch_id": str, "lot": str},  # 指定类型，防止前导0丢失
    low_memory=False   # 大文件混列时避免警告
)

注意编码问题：我曾经在读取客户提供的 CSV 时，默认 utf-8 直接报错。后来发现是 gbk 编码。建议先试 utf-8，报错就用 encoding="gbk"。还不行的，用 chardet 库自动检测。

你想想看，batch_id 如果被读成整型，'00123' 就变成 123，后续匹配数据库记录时根本对不上。所以 dtype 参数不能省。

Excel 文件读取——兼容两个版本

Excel 分 .xls（老版）和 .xlsx（新版）。pandas 的 read_excel 能自动识别，但底层依赖不同引擎。

# 读取 .xlsx
df_xlsx = pd.read_excel("qc_records.xlsx", sheet_name="Sheet1")

# 读取 .xls（老格式）
df_xls = pd.read_excel("qc_records.xls", engine="xlrd")

# 读取指定区域
df_range = pd.read_excel("qc_records.xlsx", header=0, skiprows=2, nrows=100)

小贴士：Excel 里常见合并单元格。读进来后合并区域只有第一个单元格有值，其他是 NaN。我建议用 ffill() 填充：
df["列名"] = df["列名"].ffill()

还有一个坑：同一个 Excel 文件，不同 Sheet 的列名可能不一致。我曾经遇到过 Sheet1 叫「批号」，Sheet2 叫「BatchNo」，合并时就炸了。解决方案是统一映射列名。

JSON——记录型数据的最爱

质控系统对接 REST API 时，返回的往往是 JSON。JSON 结构灵活，但转成 DataFrame 需要一点技巧。

import json
import pandas as pd

# 示例 JSON
data = [
    {"batch_id": "B001", "measurements": {"temp": 25.1, "ph": 7.2}},
    {"batch_id": "B002", "measurements": {"temp": 25.3, "ph": 7.0}}
]

# 方法二：json_normalize 展开嵌套
df = pd.json_normalize(data)
print(df.head())

# 输出示例
#   batch_id  measurements.temp  measurements.ph
# 0    B001               25.1              7.2
# 1    B002               25.3              7.0

重点：记录型的 JSON（数组里套对象）用 pd.json_normalize。如果 JSON 是对象嵌套对象，用 pandas.read_json 或者自己写递归展开。

说白了，JSON 转 DataFrame 的关键就是搞清楚「行」在哪里。数组的元素就是行，对象的 key 是列。嵌套字段用点号展开。

数据库表读取——生产环境的主战场

大部分工厂的质控数据都存在数据库里。我遇到过 SQLite、MySQL、PostgreSQL、SQL Server。别怕，pandas 的 read_sql 把差异都封装好了。

先建立连接，再执行查询：

from sqlalchemy import create_engine
import pandas as pd

# SQLite 示例（最轻量，适合原型验证）
engine = create_engine("sqlite:///qc_2025.db")

# MySQL 示例
# engine = create_engine("mysql+pymysql://user:pass@host:3306/qc_db")

# PostgreSQL 示例
# engine = create_engine("postgresql://user:pass@host:5432/qc_db")

# 读取整表
df = pd.read_sql("SELECT * FROM batch_qc WHERE date >= '2025-01-01'", engine)

# 分块读取大表（1000条一批，防止内存炸掉）
chunks = []
for chunk in pd.read_sql("SELECT * FROM batch_qc", engine, chunksize=1000):
    chunks.append(chunk)
df = pd.concat(chunks)

数据库连接安全：千万不要把用户名密码写死在代码里。我习惯用环境变量或配置文件：
import os
DB_URL = os.getenv("QC_DB_URL")

还有一个实用技巧：SQL 里尽量做「列修剪」。不要 SELECT *，只取你需要的列，能省下大量内存和网络开销。

数据源接入对比表

数据源	推荐接口	关键参数	读取方式	适用场景
CSV	pd.read_csv()	sep, encoding, dtype	一次性 / 分块	仪器导出、历史数据
Excel	pd.read_excel()	sheet_name, engine, skiprows	一次性	手工录入、客户发来的报表
JSON	pd.json_normalize()	record_path, meta	一次性	API 接口、物联网设备上报
SQL数据库	pd.read_sql()	sql, con, chunksize	一次性 / 分块	生产系统、MES、LIMS

我的经验：做数据接入时，先写一个「适配器」函数。每种数据源一个 handler，返回统一的 DataFrame。这样后续质控逻辑就不需要关心数据从哪里来。我管这叫「来源无关化」。

嗯，举个例子：

def load_data(source_type, path_or_query, **kwargs):
    if source_type == "csv":
        return pd.read_csv(path_or_query, **kwargs)
    elif source_type == "excel":
        return pd.read_excel(path_or_query, **kwargs)
    elif source_type == "json":
        with open(path_or_query) as f:
            data = json.load(f)
        return pd.json_normalize(data)
    elif source_type == "sql":
        engine = create_engine(kwargs.get("db_url"))
        return pd.read_sql(path_or_query, engine)
    else:
        raise ValueError(f"不支持的数据源类型: {source_type}")

# 统一调用
df1 = load_data("csv", "data.csv", encoding="gbk")
df2 = load_data("sql", "SELECT * FROM qc", db_url="sqlite:///qc.db")

你想想看，有了这个适配器，以后新增数据源只需要加一个 elif 分支。维护起来省心多了。

最后提醒一句：无论从哪种数据源读数据，读完后第一件事不是分析，而是校验。检查行数、列数、缺失值。我曾经因为读到一个空文件，后续计算全崩了，排查了一下午才找到根因。