自学教程 学编程从这里开始

Pandas 读写 SQLite

Pandas 教程

说明：SQLite 是文件型轻量数据库，无需单独启动服务，单文件即可；Pandas 通过 sqlite3 / sqlalchemy 实现读写。适合中小规模数据集、本地持久化、小项目数据存储。
推荐优先使用 SQLAlchemy，参数更友好，支持多种数据库方言；原生 sqlite3 适合简单场景。

1. 环境安装

# sqlite3 是 Python 内置，无需额外安装
# 如需使用 sqlalchemy（推荐）
pip install pandas sqlalchemy

2. 基础语法

2.1 to_sql 写入

df.to_sql(
    name,                # 表名
    con,                 # 数据库连接对象
    if_exists="fail",    # fail / replace / append
    index=True,          # 是否写入索引列
    index_label=None,
    dtype=None,          # 指定字段类型
    chunksize=None       # 分批写入行数，大数据防内存溢出
)
  • if_exists="fail"：表已存在直接报错（默认）
  • if_exists="replace"：先删表再重建，写入新数据
  • if_exists="append"：追加数据到已有表

2.2 read_sql / read_sql_query 读取

pd.read_sql(
    sql,                 # SQL语句 或者 表名
    con,
    index_col=None,
    params=None,         # SQL参数绑定，防注入
    chunksize=None       # 分批读取
)

read_sql_query：专门执行查询语句；read_sql_table：直接按表名读取整张表；read_sql 兼容两者。

3. 方案一：内置 sqlite3（极简）

import pandas as pd
import sqlite3

# 构造测试数据
df = pd.DataFrame({
    "id": [1, 2, 3, 4, 5],
    "name": ["张三", "李四", "王五", "赵六", "钱七"],
    "score": [88, 92, 76, 81, 95],
    "create_time": pd.date_range("2026-01-01", periods=5)
})

# 连接数据库，不存在则自动创建 test.db
conn = sqlite3.connect("test.db")

# 写入表 student
df.to_sql(
    name="student",
    con=conn,
    if_exists="replace",
    index=False
)

# 读取整张表
df_read = pd.read_sql("student", conn)
print(df_read)

# 执行条件查询
df_query = pd.read_sql("SELECT * FROM student WHERE score > 80", conn)
print("\n分数大于80：")
print(df_query)

conn.close()

4. 方案二：SQLAlchemy（推荐，工程化）

import pandas as pd
from sqlalchemy import create_engine

# 创建连接引擎，sqlite:/// 后面是本地文件路径
engine = create_engine("sqlite:///test.db")

df = pd.DataFrame({
    "id": [1, 2, 3],
    "name": ["A", "B", "C"],
    "val": [10, 20, 30]
})

# 写入
df.to_sql("demo_table", con=engine, if_exists="replace", index=False)

# 查询
df_res = pd.read_sql_query("SELECT * FROM demo_table WHERE val > 10", con=engine)
print(df_res)

5. 重要进阶用法

5.1 参数化查询（防止SQL注入，必须养成习惯）

不要直接字符串拼接 SQL！使用 params

engine = create_engine("sqlite:///test.db")
sql = "SELECT * FROM student WHERE score > ?"
df = pd.read_sql(sql, con=engine, params=(85,))
print(df)

5.2 分批写入 chunksize（大数据量）

数据量很大，一次性写入容易卡顿：

# 每批次 1000 行写入
df.to_sql("student", con=engine, if_exists="append", index=False, chunksize=1000)

5.3 分批读取（流式读取，不一次性加载全表）

返回迭代器，循环处理：

engine = create_engine("sqlite:///test.db")
reader = pd.read_sql("SELECT * FROM student", con=engine, chunksize=2)
for chunk_df in reader:
    print("批次：")
    print(chunk_df)

5.4 指定字段类型 dtype

sqlite 默认类型推断有时不符合预期，手动指定：

from sqlalchemy.types import Integer, String, DateTime

dtype_map = {
    "id": Integer(),
    "name": String(50),
    "score": Integer(),
    "create_time": DateTime()
}
df.to_sql("student", con=engine, if_exists="replace", index=False, dtype=dtype_map)

5.5 事务控制（sqlite3原生）

多条写入需要一起成功/回滚：

import pandas as pd
import sqlite3

conn = sqlite3.connect("test.db")
try:
    df1.to_sql("t1", conn, if_exists="append", index=False)
    df2.to_sql("t2", conn, if_exists="append", index=False)
    conn.commit()
    print("提交成功")
except Exception as e:
    conn.rollback()
    print("回滚，错误：", e)
conn.close()

6. 常用 DDL 配合（建索引、查看表）

engine = create_engine("sqlite:///test.db")
with engine.connect() as c:
    # 创建索引，加快查询
    c.execute("CREATE INDEX IF NOT EXISTS idx_student_id ON student(id)")
    # 查询库内所有表
    tables = c.execute("SELECT name FROM sqlite_master WHERE type='table';").fetchall()
    print(tables)

7. 常见坑点

  1. 日期时间类型

    sqlite 本身没有 datetime，存储时会转为字符串；pandas 读回来默认是 object，需要手动转换：
df["create_time"] = pd.to_datetime(df["create_time"])

  1. if_exists append 时字段不一致报错

    追加写入时，新df列名、列顺序、类型必须和数据库表一致；replace 会重建表，不存在这个问题。

  2. sqlite 并发锁

    SQLite 文件锁机制，多进程同时写入容易报 database is locked。适合单进程场景；高并发场景不要用 SQLite，换 MySQL。

  3. 主键自增

    to_sql 不会自动创建主键。需要主键要手动执行 CREATE TABLE 语句。

CREATE TABLE student (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    score INTEGER
);
  1. 空值：pandas NaN 写入 sqlite 变成 NULL，读取回来是 NaN，兼容良好。

8. 完整实战模板：清洗数据入库 + 导出

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///business.db")

# 读取csv清洗
df = pd.read_csv("data.csv", encoding="utf-8-sig")
df = df.dropna(subset=["id"])
df["amount"] = df["amount"].astype(float)

# 入库
df.to_sql("order_data", con=engine, if_exists="replace", index=False, chunksize=2000)

# 查询并导出Excel
sql = "SELECT id,amount FROM order_data WHERE amount > 100"
df_out = pd.read_sql(sql, engine)
df_out.to_excel("sqlite_result.xlsx", index=False)
print("完成")

9. SQLite 适用边界总结

✅ 适合：本地脚本、单机小工具、测试、中小数据集、离线分析、不需要独立数据库服务
❌ 不适合：多服务并发写入、高QPS、大量用户访问、超过百万级高频写入场景

标签：

0 条笔记