我的个人技术笔记

4.6 SQL

4 min read

pandas 可以通过 SQLAlchemy 引擎连接多种数据库（SQLite、MySQL、PostgreSQL、SQL Server、Oracle 等），或将 pandas 数据写入数据库表。此外也支持 Python 内置的 sqlite3 直接连接 SQLite 数据库。

1. 环境准备

import pandas as pd
from sqlalchemy import create_engine
 
# SQLite（无需安装额外的驱动）
engine = create_engine('sqlite:///mydb.db')
 
# MySQL
# engine = create_engine('mysql+pymysql://user:password@host:port/db')
# PostgreSQL
# engine = create_engine('postgresql+psycopg2://user:password@host:port/db')
# SQL Server
# engine = create_engine('mssql+pyodbc://user:password@host:port/db')
# Oracle
# engine = create_engine('oracle+cx_oracle://user:password@host:port/db')

依赖提示

  • SQLAlchemy：pip install sqlalchemy
  • 各数据库驱动：pymysql、psycopg2-binary、pyodbc、cx_oracle 等
  • SQLite 可通过 sqlite3 直接使用，无需安装额外依赖

2. read_sql()

读取 SQL 查询结果或整张表，是 read_sql_query 与 read_sql_table 的统一入口。

pd.read_sql(
    sql, con, index_col=None, coerce_float=True,
    params=None, parse_dates=None, columns=None,
    chunksize=None, dtype=None, dtype_backend=_NoDefault.no_default
)
参数说明
sqlSQL 查询字符串或表名
conSQLAlchemy 引擎、连接对象或 sqlite3.Connection
index_col作为索引的列
coerce_float是否将数值字符串转为浮点
paramsSQL 参数（字典或列表），防止 SQL 注入
parse_dates日期解析列
columns读取列（当 sql 为表名时）
chunksize分块读取行数
dtype列类型映射
dtype_backend类型后端
# 直接查询
df = pd.read_sql('SELECT * FROM employees', engine)
 
# 带参数查询（安全）
df = pd.read_sql(
    'SELECT * FROM employees WHERE department = :dept',
    engine,
    params={'dept': 'Sales'}
)
 
# 读取整张表
df = pd.read_sql('employees', engine)
 
# 分块读取
chunks = pd.read_sql('SELECT * FROM big_table', engine, chunksize=10000)

3. read_sql_query()

只用于执行 SQL 查询语句，与 read_sql 的区别是参数不接受表名。

pd.read_sql_query(
    sql, con, index_col=None, coerce_float=True, params=None,
    parse_dates=None, chunksize=None, dtype=None, dtype_backend=_NoDefault.no_default
)
df = pd.read_sql_query('SELECT id, name FROM users WHERE age > ?', sqlite_conn, params=(25,))

4. read_sql_table()

只用于读取整张数据库表，不执行 SQL 查询。需要 SQLAlchemy 引擎，且需指定表名。

pd.read_sql_table(
    table_name, con, schema=None, index_col=None,
    coerce_float=True, parse_dates=None, columns=None,
    chunksize=None, dtype=None, dtype_backend=_NoDefault.no_default
)
参数说明
table_name表名
conSQLAlchemy 引擎（必须）
schema数据库模式（Schema）
columns读取列
# 读取整表
df = pd.read_sql_table('employees', engine)
 
# 指定 schema 与列
df = pd.read_sql_table('employees', engine, schema='hr', columns=['id', 'name'])

5. to_sql()

将 DataFrame 写入数据库表。

DataFrame.to_sql(
    name, con, schema=None, if_exists='fail', index=True,
    index_label=None, chunksize=None, dtype=None, method=None
)

参数详解

参数说明
name目标表名
conSQLAlchemy 引擎或 sqlite3.Connection
schema目标 schema
if_exists'fail'（已存在则报错）、'replace'（替换）、'append'（追加）
index是否写入索引
index_label索引的列名
chunksize分块写入大小
dtype列类型映射（SQLAlchemy 类型）
method写入方法：None、'multi' 或自定义可调用对象
# 写入新表
df.to_sql('sales', engine, if_exists='replace', index=False)
 
# 追加数据
df.to_sql('sales', engine, if_exists='append', index=False)
 
# 分块写入
df.to_sql('big_table', engine, if_exists='append', chunksize=5000)
 
# 指定数据库列类型
from sqlalchemy.types import Integer, String
df.to_sql(
    'my_table', engine,
    dtype={
        'id': Integer,
        'name': String(50)
    }
)

sqlite3

Python 内置的 sqlite3 模块无需额外安装，可直接作为 con 参数使用。

import sqlite3
 
# 连接数据库（不存在则自动创建）
conn = sqlite3.connect('mydb.db')
 
# 读取
df = pd.read_sql('SELECT * FROM employees', conn)
 
# 写入
df.to_sql('employees', conn, if_exists='replace', index=False)
 
# 关闭
conn.close()

建议

  • 所有 SQL 查询应使用 params 传参，避免 SQL 注入。
  • read_sql_table 只支持 SQLAlchemy，sqlite3 连接可用 read_sql / read_sql_query。
  • 大数据量写入时建议设置 chunksize，减少内存压力。

文档 8：4.7 其他数据格式.md

Graph View

Backlinks