【零基础玩转Python】Day55：数据读写 – CSV, Excel, JSON, SQL连接，打通数据获取的任督二脉

大家好，我是[知识充电宝的灵感日记]。

在数据分析流程中，第一步往往是读取数据——可能来自 CSV 文件、Excel 表格、JSON 接口或数据库。而分析完成后，我们也需要将结果保存到文件或数据库。今天我们就来系统学习 Pandas 如何与这些常见数据源交互。

今天的目标：

• ✅ 读取和写入 CSV 文件（最常用）
• ✅ 读取和写入 Excel 文件（多工作表）
• ✅ 读取和写入 JSON 文件（嵌套结构处理）
• ✅ 连接 SQL 数据库（SQLite、MySQL 等）并执行查询
• ✅ 处理不同编码、分隔符、缺失值标记等问题
• ✅ 实战：从多种数据源整合数据

难度：⭐⭐（函数参数较多，但规律性强）

一、CSV 文件的读写

CSV（Comma-Separated Values）是最通用的表格数据格式。

1. 读取 CSV：pd.read_csv()

import pandas as pd

# 基本读取
df = pd.read_csv('data.csv')
print(df.head())

# 常用参数
df = pd.read_csv('data.csv',
                 encoding='utf-8',           # 编码，常见 utf-8, gbk
                 sep=',',                     # 分隔符，默认逗号
                 header=0,                    # 哪一行作为列名（0为第一行）
                 index_col=0,                 # 将第0列作为行索引
                 usecols=['col1', 'col2'],    # 只读取指定列
                 dtype={'col1': str},         # 指定列的数据类型
                 parse_dates=['date_col'],    # 将列解析为日期
                 na_values=['', 'NA', 'null'],# 将指定值识别为缺失值
                 nrows=100,                   # 只读取前100行
                 skiprows=[0,2],              # 跳过指定行
                 thousands=',',               # 千分位分隔符（如"1,000"）
                 )

2. 写入 CSV：df.to_csv()

df.to_csv('output.csv', 
          index=False,          # 不写入行索引
          encoding='utf-8',
          sep=',',
          columns=['col1','col2'],  # 只写入指定列
          header=True,          # 是否写入列名
          na_rep='NULL'# 缺失值填充字符串
          )

二、Excel 文件的读写

需要安装 openpyxl（用于 .xlsx）或 xlrd（用于旧版 .xls）：

pip install openpyxl xlrd

1. 读取 Excel：pd.read_excel()

# 读取第一个工作表
df = pd.read_excel('data.xlsx')

# 指定工作表名或索引
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')
df = pd.read_excel('data.xlsx', sheet_name=0)   # 第一个工作表

# 读取多个工作表
sheets = pd.read_excel('data.xlsx', sheet_name=['Sheet1', 'Sheet2'])
# 或 sheet_name=None 读取所有工作表，返回字典
all_sheets = pd.read_excel('data.xlsx', sheet_name=None)

其他参数（如 header, skiprows等）与 read_csv 类似。

2. 写入 Excel：df.to_excel()

# 写入单个工作表
df.to_excel('output.xlsx', sheet_name='Sheet1', 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)

三、JSON 文件的读写

JSON 是半结构化数据，常用于 API 接口。

1. 读取 JSON：pd.read_json()

# 从文件读取
df = pd.read_json('data.json')

# 从 JSON 字符串读取
json_str = '[{"name": "A", "value": 1}, {"name": "B", "value": 2}]'
df = pd.read_json(json_str)

# 处理嵌套 JSON（需要用 json_normalize）
import json
withopen('nested.json') as f:
    data = json.load(f)
df = pd.json_normalize(data, record_path='records', meta=['meta_field'])

2. 写入 JSON：df.to_json()

df.to_json('output.json', orient='records', force_ascii=False, indent=2)

• orient：输出格式，常用 'records'（列表形式），'index'（键为索引），'columns'（键为列名）
• force_ascii=False：保留中文
• indent：美化缩进

四、SQL 数据库连接

Pandas 可以通过 sqlalchemy 连接各种数据库（SQLite, MySQL, PostgreSQL 等）。

安装依赖：

pip install sqlalchemy pymysql   # MySQL 驱动
# 或 pip install sqlalchemy psycopg2  # PostgreSQL

1. 创建数据库连接

from sqlalchemy import create_engine

# SQLite（文件数据库）
engine = create_engine('sqlite:///database.db')

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

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

2. 读取 SQL 查询结果：pd.read_sql()

# 方式1：直接写 SQL 查询
df = pd.read_sql('SELECT * FROM orders WHERE amount > 100', engine)

# 方式2：读取整个表
df = pd.read_sql_table('orders', engine)

# 方式3：查询 + 参数（防注入）
df = pd.read_sql_query('SELECT * FROM orders WHERE date >= :start_date', 
                       engine, params={'start_date': '2025-01-01'})

3. 写入数据库：df.to_sql()

df.to_sql('orders', 
          con=engine, 
          if_exists='replace',   # 'fail', 'replace', 'append'
          index=False,
          chunksize=1000)        # 分批写入，避免内存过大

if_exists 选项：

• 'fail'：表已存在时抛出错误
• 'replace'：删除原表并重建
• 'append'：追加数据

4. 完整示例：SQLite 数据库操作

import sqlite3
import pandas as pd

# 创建 SQLite 连接（内存数据库或文件）
conn = sqlite3.connect('sales.db')

# 创建表并插入示例数据
df_sample = pd.DataFrame({
'product': ['A', 'B', 'C'],
'sales': [100, 200, 150]
})
df_sample.to_sql('products', conn, if_exists='replace', index=False)

# 查询数据
df = pd.read_sql('SELECT * FROM products WHERE sales > 120', conn)
print(df)

conn.close()

五、实战小案例：多源数据整合

假设我们有三个数据源：

• 用户信息（CSV）
• 订单明细（Excel）
• 产品类别映射（JSON）

我们将它们读取并合并为统一的分析数据集。

# 1. 读取 CSV
users = pd.read_csv('users.csv', encoding='utf-8')
print("用户数据形状:", users.shape)

# 2. 读取 Excel 订单数据
orders = pd.read_excel('orders.xlsx', sheet_name='OrderDetails')
print("订单数据形状:", orders.shape)

# 3. 读取 JSON 产品映射
withopen('products.json', 'r', encoding='utf-8') as f:
    product_map = pd.json_normalize(json.load(f), 'products')
print("产品映射形状:", product_map.shape)

# 4. 数据合并
merged = orders.merge(users, on='user_id', how='left')
merged = merged.merge(product_map, on='product_id', how='left')
print("合并后形状:", merged.shape)

# 5. 分析结果保存到数据库
from sqlalchemy import create_engine
engine = create_engine('sqlite:///analysis.db')
merged.to_sql('analysis_result', engine, if_exists='replace', index=False)
print("结果已保存到数据库")

六、今日练习

1. 从本地或网络下载一个 CSV 文件（如泰坦尼克数据集 titanic.csv），用 pd.read_csv读取，并尝试使用不同的参数（如 encoding, na_values）。
2. 创建一个简单的 DataFrame，用 to_excel保存为 Excel 文件，并包含两个工作表。
3. 用 Python 的 requests库获取一个公开 API 的 JSON 数据（如 https://api.github.com/repos/pandas-dev/pandas），解析为 DataFrame。
4. （挑战）使用 SQLite 内存数据库，创建一个 employees表（字段：id, name, salary），插入 5 条数据，然后用 pd.read_sql查询出工资大于 5000 的员工。

七、常见错误与提示

• CSV 编码问题：常用 encoding='utf-8'或 gbk，可尝试 encoding='utf-8-sig'处理 BOM 头。
• Excel 引擎：.xlsx用 openpyxl，.xls用 xlrd，需提前安装。
• JSON 嵌套结构：pd.json_normalize可以扁平化嵌套字典/列表。
• 数据库连接安全：避免在代码中硬编码数据库密码，可使用环境变量或配置文件。
• 大数据写入数据库：使用 chunksize分批写入，避免内存溢出。
• pd.read_sql依赖 SQLAlchemy：也可以直接使用 sqlite3模块的 cursor，但 Pandas 推荐使用 SQLAlchemy。

八、明日预告

Day56：实战小项目 – 用Pandas分析用户活跃度

我们将综合运用数据处理和可视化技能，分析用户行为数据（如登录次数、活跃时长），给出用户分层建议。