3. 数据读取与写入：CSV、Excel、JSON、数据库等常见数据源的读取与写入

数据清洗的第一步是什么？说白了，你得先把数据弄进来。

我见过太多新手，花了一整天写清洗逻辑，结果发现数据读进来就是错的。嗯，这很常见。所以今天咱们就聊聊，怎么把各种常见的数据源——CSV、Excel、JSON、数据库——稳稳当当地读进来，再干干净净地写出去。

3.1 CSV 文件的读取与写入

CSV 是最常见的数据交换格式。我个人习惯用 pandas.read_csv()，但有几个坑你得注意。

3.1.1 基本读取

import pandas as pd

# 最基础的读取
df = pd.read_csv('data.csv')

# 指定编码，避免乱码
df = pd.read_csv('data.csv', encoding='utf-8')

# 如果文件是 GBK 编码（国内很多老系统用）
df = pd.read_csv('data.csv', encoding='gbk')

3.1.2 处理常见问题

• 分隔符不是逗号：有些系统用分号或制表符。用 sep=';' 或 sep='\t' 指定。
• 表头问题：如果文件没有表头，设置 header=None，再手动指定列名 names=['col1', 'col2']。
• 数据类型推断：pandas 会自动推断类型，但有时会出错。比如身份证号被当成数字。用 dtype={'id': str} 强制指定。

# 处理没有表头的文件
df = pd.read_csv('data_no_header.csv', header=None, names=['id', 'name', 'age'])

# 强制指定列类型
df = pd.read_csv('data.csv', dtype={'user_id': str, 'amount': float})

3.1.3 写入 CSV

# 基本写入
df.to_csv('output.csv', index=False)

# 指定编码，避免中文乱码
df.to_csv('output.csv', index=False, encoding='utf-8-sig')

3.2 Excel 文件的读取与写入

Excel 文件比 CSV 复杂，因为它可以有多张工作表。我建议用 pandas 配合 openpyxl 或 xlrd 引擎。

3.2.1 读取 Excel

# 读取默认工作表（第一张）
df = pd.read_excel('data.xlsx')

# 读取指定工作表
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')

# 读取所有工作表，返回字典
dfs = pd.read_excel('data.xlsx', sheet_name=None)
# dfs 是一个字典，键是工作表名，值是 DataFrame

3.2.2 处理合并单元格

Excel 里经常有合并单元格。pandas 读进来后，合并的单元格只有第一个有值，其他是 NaN。你需要手动填充。

# 读取时保留合并单元格的样式
df = pd.read_excel('data.xlsx', header=None)

# 向前填充合并单元格的值
df = df.fillna(method='ffill')

3.2.3 写入 Excel

# 写入单个工作表
df.to_excel('output.xlsx', sheet_name='清洗结果', index=False)

# 写入多个工作表
with pd.ExcelWriter('output.xlsx') as writer:
    df1.to_excel(writer, sheet_name='原始数据', index=False)
    df2.to_excel(writer, sheet_name='清洗后', index=False)

3.3 JSON 文件的读取与写入

JSON 在 API 接口和 NoSQL 数据库中很常见。它的结构比 CSV 灵活，但也更复杂。

3.3.1 读取 JSON

# 读取标准 JSON 文件
df = pd.read_json('data.json')

# 读取嵌套 JSON（需要指定路径）
df = pd.read_json('data.json', orient='records')

3.3.2 处理嵌套 JSON

JSON 经常嵌套多层。比如一个用户信息里包含地址对象。你需要用 json_normalize 展开。

from pandas import json_normalize

# 假设 data 是嵌套的 JSON 列表
data = [
    {"name": "张三", "address": {"city": "北京", "district": "海淀"}},
    {"name": "李四", "address": {"city": "上海", "district": "浦东"}}
]

# 展开嵌套字段
df = json_normalize(data)
# 结果：name, address.city, address.district

3.3.3 写入 JSON

# 写入为 JSON 文件
df.to_json('output.json', orient='records', force_ascii=False)

3.4 数据库的读取与写入

数据库读取是数据工程师的日常。我主要用 SQLAlchemy 连接各种数据库。

3.4.1 连接数据库

from sqlalchemy import create_engine

# MySQL 连接
engine = create_engine('mysql+pymysql://user:password@host:port/database')

# PostgreSQL 连接
engine = create_engine('postgresql://user:password@host:port/database')

# SQLite 连接（本地文件）
engine = create_engine('sqlite:///data.db')

3.4.2 读取数据

# 读取整张表
df = pd.read_sql_table('table_name', engine)

# 使用 SQL 查询
df = pd.read_sql_query('SELECT * FROM table_name WHERE date > "2024-01-01"', engine)

# 更推荐的方式：分块读取，避免内存溢出
chunks = pd.read_sql_query('SELECT * FROM large_table', engine, chunksize=10000)
for chunk in chunks:
    # 处理每一块数据
    process(chunk)

3.4.3 写入数据

# 写入数据库
df.to_sql('table_name', engine, if_exists='replace', index=False)

# 追加写入
df.to_sql('table_name', engine, if_exists='append', index=False)

# 批量写入（大数据量时推荐）
df.to_sql('table_name', engine, if_exists='append', index=False, chunksize=10000)

3.5 知识体系总览

下面这张图，是我自己总结的数据读取与写入的核心逻辑。你想想看，不管数据源是什么，流程其实都一样：连接 -> 读取 -> 处理 -> 写入。

3.6 实战建议

说了这么多，我给你几个实战建议：

1. 先检查再处理：读入数据后，先用 df.info() 和 df.head() 看一眼。数据类型对不对？有没有空值？
2. 统一编码：所有文件统一用 UTF-8 编码。如果遇到乱码，试试 GBK 或 GB2312。
3. 分块处理大文件：超过 1GB 的文件，别一次性读入内存。用 chunksize 分块处理。
4. 备份原始数据：写入前，先备份原始数据。万一写错了，还能恢复。

---