数据库数据读取（SQL） - 数据分析-学习笔记-上海佼艾

一、Python数据库交互概述

在数据分析工作流中，数据通常存储在关系数据库中。Python提供了多种方式来连接和操作数据库，其中Pandas + SQLAlchemy组合是最主流、最优雅的方案。Pandas的read_sql系列函数可以直接将SQL查询结果读取为DataFrame，而to_sql可以将DataFrame写入数据库表，极大简化了数据管道开发。

Python数据库交互的核心库包括：sqlite3（Python内置，轻量级）、SQLAlchemy（ORM框架，支持多种数据库）、psycopg2（PostgreSQL驱动）、pymysql（MySQL驱动）。在实际数据分析中，推荐直接使用SQLAlchemy作为统一接口，它可以屏蔽底层数据库差异，一套代码兼容SQLite、MySQL、PostgreSQL等多种数据库。

核心原则：数据分析场景下优先使用Pandas的数据库I/O函数（read_sql/to_sql），它们自动处理类型转换和内存管理，让分析者专注于业务逻辑而非数据库底层细节。

1.1 准备工作环境

开始之前需要安装必要的库。推荐使用以下命令进行安装：

# 安装核心依赖
pip install pandas sqlalchemy

# 根据数据库选择安装对应驱动
pip install pymysql       # MySQL
pip install psycopg2      # PostgreSQL
pip install psycopg2-binary  # PostgreSQL（简化版）

1.2 导入库

import pandas as pd
from sqlalchemy import create_engine, text
import sqlalchemy

二、SQLAlchemy数据库连接

SQLAlchemy是Python生态系统中最强大的数据库工具包之一，它提供了连接池管理、SQL方言自动适配、ORM映射等功能。在数据读取场景中，我们主要使用它的Engine对象作为数据库连接的核心入口。

2.1 连接字符串格式

SQLAlchemy使用统一的连接字符串（Connection URL）格式来标识数据库连接信息：

# 标准连接字符串格式
dialect+driver://username:password@host:port/database

# 各部分含义
# dialect  : 数据库类型，如 mysql、postgresql、sqlite
# driver   : 数据库驱动，如 pymysql、psycopg2（可省略，使用默认）
# username : 数据库用户名
# password : 数据库密码
# host     : 数据库主机地址
# port     : 数据库端口
# database : 数据库名称

2.2 各数据库连接示例

SQLite（最简，无需密码和主机）

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

# SQLite 内存数据库
engine = create_engine('sqlite:///:memory:')

# Windows 绝对路径
engine = create_engine('sqlite:///C:\\path\\to\\data.db')

MySQL

# MySQL 连接（使用 pymysql 驱动）
engine = create_engine(
    'mysql+pymysql://root:password@localhost:3306/mydb',
    echo=False,        # 是否打印SQL日志
    pool_size=5,        # 连接池大小
    max_overflow=10      # 最大溢出连接数
)

PostgreSQL

# PostgreSQL 连接（使用 psycopg2 驱动）
engine = create_engine(
    'postgresql+psycopg2://user:password@localhost:5432/mydb',
    pool_size=5,
    max_overflow=10
)

# 或者使用 psycopg2-binary（开发环境推荐）
engine = create_engine(
    'postgresql+psycopg2://user:password@localhost:5432/mydb'
)

2.3 create_engine常用参数

参数	说明	默认值
echo	是否打印执行的SQL语句（调试用）	False
pool_size	连接池中保持的连接数	5
max_overflow	连接池最大溢出连接数	10
pool_recycle	连接回收时间（秒），防止连接超时	-1（不过期）
pool_pre_ping	每次使用连接前检查是否存活	False
connect_args	传递给底层驱动的额外参数（字典）	{}

最佳实践：生产环境建议设置 pool_pre_ping=True 和 pool_recycle=3600，避免因长时间空闲导致连接被数据库服务端断开。

2.4 连接管理

SQLAlchemy Engine对象本身是线程安全的，推荐在应用生命周期内复用同一个Engine实例，而不是每次都创建新的连接。获取实际数据库连接的方式有两种：

方式一：直接使用 Engine.connect()

# 使用上下文管理器自动管理连接生命周期
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users"))
    for row in result:
        print(row)

# 连接会在退出 with 块时自动归还到连接池

方式二：使用 Engine.begin() 事务管理

# 事务管理，自动提交或回滚
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users VALUES (:id, :name)"),
                 [{"id": 1, "name": "Alice"},
                  {"id": 2, "name": "Bob"}])
# 事务在退出时自动提交，发生异常则自动回滚

注意：永远不要直接调用 engine.dispose() 除非你确定不再使用该引擎。每次调用 dispose 会关闭所有连接池中的连接，下次查询需要重新建立连接，影响性能。

2.5 连接池工作机制

连接池是SQLAlchemy性能的关键。其工作流程如下：

初始化：Engine创建时，连接池为空
请求连接：调用 engine.connect() 时，从池中获取空闲连接
创建新连接：如果池中无空闲连接且未达到 pool_size，创建新连接
等待/溢出：如果池满且所有连接都在使用，等待空闲或创建溢出连接（max_overflow）
归还连接：连接关闭时归还到池中，供后续复用
连接回收：超过 pool_recycle 时间的连接会被自动关闭并重建

三、Pandas读取数据库

Pandas提供了三个主要的数据库读取函数，它们都接受SQLAlchemy Engine或Connection对象作为连接参数：

函数	功能	适用场景
pd.read_sql()	通用读取接口，自动判断SQL还是表名	大多数情况，推荐使用
pd.read_sql_query()	执行SQL查询语句	明确需要执行SQL时
pd.read_sql_table()	读取整张数据库表	需要读取整张表时

3.1 read_sql 通用接口

pd.read_sql() 是最高层级的封装，它根据传入的第一个参数自动判断是SQL语句还是表名：

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine('sqlite:///sales.db')

# 方式一：传入SQL查询语句
df = pd.read_sql("SELECT * FROM orders WHERE amount > 100", engine)
print(df.head())

# 方式二：传入表名（需要指定 columns 参数）
df = pd.read_sql('orders', engine, columns=['id', 'amount', 'date'])
print(df.head())

3.2 read_sql_query 执行SQL

当明确需要执行SQL查询时使用，支持复杂SQL包括JOIN、子查询、窗口函数等：

# 多表关联查询
sql = """
SELECT
    o.order_id,
    o.order_date,
    c.customer_name,
    p.product_name,
    oi.quantity,
    oi.unit_price,
    oi.quantity * oi.unit_price AS total_amount
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
JOIN order_items oi ON o.order_id = oi.order_id
JOIN products p ON oi.product_id = p.product_id
WHERE o.order_date >= '2025-01-01'
ORDER BY o.order_date DESC
"""

df_orders = pd.read_sql_query(sql, engine)
print(f"共读取 {len(df_orders)} 条订单明细")
print(df_orders.describe())

3.3 read_sql_table 读取整表

专门用于读取整张数据库表，支持指定列名和行范围：

# 读取整张表
df_users = pd.read_sql_table('users', engine)

# 指定列和行范围
df_users = pd.read_sql_table(
    'users',
    engine,
    columns=['id', 'name', 'email', 'created_at'],
    schema='public'
)

# 注意：read_sql_table 不支持 WHERE 条件过滤
# 需要过滤时请使用 read_sql_query

3.4 参数传递

在实际项目中，直接拼接SQL字符串存在SQL注入风险。推荐使用参数化查询：

# 方式一：使用 :name 占位符（SQLAlchemy风格）
df = pd.read_sql_query(
    "SELECT * FROM orders WHERE amount > :min_amount AND status = :status",
    engine,
    params={"min_amount": 1000, "status": "completed"}
)

# 方式二：使用 %s 占位符（原生驱动风格）
df = pd.read_sql_query(
    "SELECT * FROM orders WHERE amount > %s AND status = %s",
    engine,
    params=(1000, "completed")
)

# 方式三：IN 子句参数化（自动展开）
status_list = ['completed', 'shipped', 'pending']
df = pd.read_sql_query(
    "SELECT * FROM orders WHERE status IN :statuses",
    engine,
    params={"statuses": status_list}
)

安全提示：始终使用参数化查询而不是f-string拼接参数。这样做不仅防止SQL注入，还能让数据库缓存查询计划，提升重复查询性能。

3.5 read_sql 完整参数详解

参数	类型	说明
sql	str	SQL查询语句或数据库表名
con	Engine/Connection	SQLAlchemy引擎或连接对象
params	dict/list/tuple	SQL查询参数，用于参数化查询
index_col	str/list	指定作为行索引的列名
columns	list	读取表时指定列名（仅read_sql_table）
chunksize	int	分块读取，每次返回指定行数的迭代器
schema	str	数据库schema名称
parse_dates	list/dict	指定解析为日期时间的列
dtype	dict	指定列的数据类型（type name or dict）

# 综合示例：使用完整参数
df = pd.read_sql(
    "SELECT id, name, created_at, score FROM users WHERE score > :min_score",
    engine,
    params={"min_score": 80},
    index_col='id',
    parse_dates=['created_at'],
    dtype={'score': float}
)

四、Pandas写入数据库

DataFrame.to_sql() 方法可以将内存中的数据批量写入数据库表，是数据管道中ETL的加载（Load）阶段的核心操作。

4.1 to_sql 基础用法

# 准备数据
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35],
    'salary': [50000.0, 60000.0, 70000.0]
})

# 写入数据库表
df.to_sql(
    'employees',     # 表名
    engine,             # 数据库引擎
    if_exists='replace', # 表已存在时的处理方式
    index=False         # 不写入行索引
)

# 验证写入结果
result = pd.read_sql("SELECT * FROM employees", engine)
print(result)

4.2 if_exists 参数详解

if_exists 参数决定当目标表已存在时的行为，有三种可选值：

值	行为	适用场景
'fail'	表已存在时抛出 ValueError	防止意外覆盖数据，生产环境默认推荐
'replace'	删除原表并创建新表	全量刷新数据，数仓ETL中常见
'append'	将数据追加到现有表中	增量写入，日志数据持续追加

# fail：表已存在时抛出异常，防止覆盖
try:
    df.to_sql('employees', engine, if_exists='fail')
except ValueError as e:
    print(f"表已存在: {e}")

# append：追加数据，适合日志或流式数据
df_new = pd.DataFrame({
    'name': ['David', 'Eve'],
    'age': [28, 32],
    'salary': [55000.0, 65000.0]
})
df_new.to_sql('employees', engine, if_exists='append', index=False)

4.3 index 与 dtype 参数

控制是否将DataFrame的行索引写入数据库表，以及自定义列的数据类型：

# 保留索引列（会写入名为 'index' 的列）
df.to_sql('employees', engine, index=True, index_label='id')

# 指定数据库列类型（SQLAlchemy类型）
from sqlalchemy.types import VARCHAR, Integer, Float, Date

df.to_sql(
    'employees',
    engine,
    if_exists='replace',
    index=False,
    dtype={
        'name': VARCHAR(100),
        'age': Integer(),
        'salary': Float(precision=2),
        'hire_date': Date()
    }
)

4.4 chunksize 批量写入

当写入大量数据时，使用 chunksize 参数分批写入，避免单次事务过大：

# 每次写入 1000 行，避免大事务
df_large.to_sql(
    'large_table',
    engine,
    if_exists='append',
    index=False,
    chunksize=1000,
    method='multi'  # 使用多值插入语法，大幅提升性能
)

性能对比：不指定 method 时逐行 INSERT，速度较慢。设置 method='multi' 后使用 INSERT INTO ... VALUES (...), (...), (...) 批量插入，速度可提升 10-50 倍。chunksize 控制每个批次的行数，推荐 1000-5000 行。

4.5 to_sql 完整参数详解

参数	类型	说明
name	str	目标数据库表名
con	Engine/Connection	SQLAlchemy引擎或连接
schema	str	数据库schema（如MySQL的数据库名）
if_exists	str	'fail'/'replace'/'append'，默认'fail'
index	bool	是否写入DataFrame索引，默认True
index_label	str/list	索引列在数据库中的列名
chunksize	int	每次批量写入的行数
dtype	dict	列名到SQLAlchemy类型的映射
method	str/None/callable	插入方法，'multi'使用多值插入

五、大数据分块处理

当数据库表数据量巨大（数百万行甚至更多）时，一次性加载到内存会导致内存溢出。Pandas提供了分块读取机制，通过 chunksize 参数返回一个迭代器，逐块处理数据。

5.1 chunksize 迭代读取

# 分块读取大数据集，每块 10000 行
chunk_iter = pd.read_sql_query(
    "SELECT * FROM massive_table",
    engine,
    chunksize=10000
)

# 逐块处理，适合聚合计算
total_rows = 0
partial_sums = []

for chunk in chunk_iter:
    total_rows += len(chunk)
    partial_sums.append(chunk['amount'].sum())
    print(f"已处理 {total_rows} 行...")

total_sum = sum(partial_sums)
print(f"共 {total_rows} 行，总金额: {total_sum}")

5.2 分块聚合模式

大数据分析中的Map-Reduce模式可以结合分块读取实现：

# Map阶段：逐块计算统计量
from collections import defaultdict

agg_result = defaultdict(lambda: {'count': 0, 'sum': 0.0, 'min': None, 'max': None})

for chunk in pd.read_sql_query(
    "SELECT category, amount FROM sales",
    engine,
    chunksize=5000
):
    for category, group in chunk.groupby('category'):
        stats = agg_result[category]
        stats['count'] += len(group)
        stats['sum'] += group['amount'].sum()
        stats['min'] = min(stats['min'] or float('inf'), group['amount'].min())
        stats['max'] = max(stats['max'] or float('-inf'), group['amount'].max())

# Reduce阶段：汇总结果
result_df = pd.DataFrame(agg_result).T
result_df['avg'] = result_df['sum'] / result_df['count']
print(result_df)

5.3 分块写入大数据

同样，写入大数据集时也需要分块处理：

# 分块写入，避免单个事务过大
chunk_size = 5000

for start in range(0, len(large_df), chunk_size):
    end = start + chunk_size
    chunk = large_df.iloc[start:end]

    chunk.to_sql(
        'target_table',
        engine,
        if_exists='append',
        index=False,
        method='multi'
    )

    print(f"已写入 {end}/{len(large_df)} 行")

分块策略建议：chunksize值并非越大越好。经验值：SQLite建议5000-10000行/块，MySQL建议5000-20000行/块，PostgreSQL建议10000-50000行/块。过大的chunksize会导致单次查询返回过多数据，增加网络传输和内存压力。

六、数据库类型与Pandas类型映射

从数据库读取数据时，SQL数据类型会自动映射为Pandas/NumPy数据类型。理解这个映射关系有助于避免类型相关的错误：

SQL类型	SQLAlchemy类型	Pandas类型	备注
INTEGER / INT	Integer	int64	32位整数自动扩展为64位
BIGINT	BigInteger	int64	直接映射
SMALLINT	SmallInteger	int64	自动扩展为64位
FLOAT / REAL	Float	float64	双精度浮点
DOUBLE PRECISION	Float	float64	双精度浮点
NUMERIC / DECIMAL	Numeric	float64	可能丢失精度，建议使用Decimal
VARCHAR / CHAR	String	object (str)	字符串类型
TEXT	Text	object (str)	长文本
DATE	Date	datetime64[ns]	日期类型
TIMESTAMP / DATETIME	DateTime	datetime64[ns]	日期时间
BOOLEAN	Boolean	bool	布尔类型
BLOB / BINARY	LargeBinary	object (bytes)	二进制数据
JSON	JSON	object (dict/list)	PostgreSQL原生JSON

# 手动控制读取时的类型映射
df = pd.read_sql_query(
    "SELECT id, price, quantity, created_at FROM transactions",
    engine,
    parse_dates=['created_at'],       # 确保解析为日期时间
    dtype={
        'id': 'int64',
        'price': 'float32',       # 使用float32节省内存
        'quantity': 'int32'       # 使用int32节省内存
    }
)

print(df.dtypes)
print(f"内存使用: {df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB")

内存优化技巧：如果数据量很大且不需要高精度，将float64降为float32可将内存减半。同样，int64降为int32也可节省50%内存。这对于数亿行数据的处理非常关键。

七、常用数据库连接示例

7.1 SQLite 完整示例

SQLite是学习和开发阶段的绝佳选择，无需安装数据库服务，所有数据存储在一个文件中：

import pandas as pd
from sqlalchemy import create_engine
import os

# 创建数据库（文件不存在会自动创建）
db_path = os.path.join(os.getcwd(), 'example.db')
engine = create_engine(f'sqlite:///{db_path}')

# 准备示例数据
df_products = pd.DataFrame({
    'product_id': range(1, 6),
    'product_name': ['笔记本', '钢笔', '橡皮', '尺子', '书包'],
    'price': [15.0, 8.5, 2.0, 5.5, 120.0],
    'stock': [100, 200, 500, 150, 80]
})

df_products.to_sql('products', engine, if_exists='replace', index=False)
print("产品表创建成功")

# 模拟销售数据
df_sales = pd.DataFrame({
    'sale_id': range(1, 101),
    'product_id': np.random.randint(1, 6, 100),
    'quantity': np.random.randint(1, 10, 100),
    'sale_date': pd.date_range('2025-01-01', periods=100, freq='D')
})

df_sales.to_sql('sales', engine, if_exists='replace', index=False)

# 使用SQL分析数据：每个产品的销售总额
query = """
SELECT
    p.product_name,
    SUM(s.quantity) AS total_quantity,
    SUM(s.quantity * p.price) AS total_revenue
FROM sales s
JOIN products p ON s.product_id = p.product_id
GROUP BY p.product_name
ORDER BY total_revenue DESC
"""

df_analysis = pd.read_sql_query(query, engine)
print(df_analysis)

7.2 MySQL 完整示例

# MySQL连接配置
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': 'your_password',
    'database': 'analysis_db',
    'charset': 'utf8mb4'
}

# 构建连接字符串
conn_str = ("mysql+pymysql://{user}:{password}@{host}:{port}/{database}"
            "?charset={charset}").format(**DB_CONFIG)

engine = create_engine(
    conn_str,
    pool_size=5,
    pool_recycle=3600,
    pool_pre_ping=True,
    echo=False
)

# 测试连接
try:
    with engine.connect() as conn:
        result = conn.execute(text("SELECT VERSION()"))
        version = result.scalar()
        print(f"成功连接到MySQL，版本: {version}")
except Exception as e:
    print(f"连接失败: {e}")

7.3 PostgreSQL 完整示例

# PostgreSQL连接配置
engine = create_engine(
    'postgresql+psycopg2://user:password@localhost:5432/analysis_db',
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    pool_pre_ping=True,
    connect_args={
        'sslmode': 'require',       # SSL连接
        'connect_timeout': 10      # 连接超时秒数
    }
)

# 读取PostgreSQL特有的JSON字段
df = pd.read_sql_query("""
    SELECT
        id,
        data->>'name' AS name,
        data->>'email' AS email,
        created_at
    FROM users
    WHERE data->>'status' = 'active'
""", engine)

# 使用schema指定模式
df_orders = pd.read_sql_table(
    'orders',
    engine,
    schema='analytics',
    columns=['id', 'amount', 'created_at']
)

八、错误处理与最佳实践

8.1 常见错误处理

数据库操作中可能遇到各种异常，良好的错误处理机制是生产级代码的必备要素：

from sqlalchemy.exc import (
    OperationalError,
    DatabaseError,
    IntegrityError,
    TimeoutError
)

def safe_read_sql(query, engine, params=None, chunksize=None, retries=3):
    """安全的SQL读取函数，包含重试机制"""
    for attempt in range(retries):
        try:
            return pd.read_sql_query(query, engine, params=params, chunksize=chunksize)

        except OperationalError as e:
            print(f"数据库连接错误 (第 {attempt+1} 次尝试): {e}")
            if attempt < retries - 1:
                import time
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise

        except DatabaseError as e:
            print(f"数据库错误: {e}")
            raise

        except TimeoutError as e:
            print(f"查询超时 (第 {attempt+1} 次尝试): {e}")
            if attempt < retries - 1:
                time.sleep(5)
            else:
                raise

    return None

# 使用示例
try:
    df = safe_read_sql("SELECT * FROM large_table WHERE date > :dt",
                       engine,
                       params={"dt": "2025-01-01"})
    if df is not None:
        print(f"成功读取 {len(df)} 行数据")
except Exception as e:
    print(f"数据读取完全失败: {e}")

8.2 连接池管理最佳实践

# 在应用启动时创建引擎，全局复用
engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    pool_use_lifo=True  # 后进先出，减少连接老化
)

# 不要在每个函数中创建新的Engine
# 错误做法：
# def get_data():
#     engine = create_engine(...)  # 每次调用都创建新引擎，浪费资源
#     return pd.read_sql("SELECT * FROM t", engine)

# 正确做法：将engine作为模块级别变量或依赖注入

8.3 性能优化清单

数据库读取性能优化要点：

只取需要的列：使用 SELECT col1, col2 而非 SELECT *
利用数据库过滤：在SQL中使用 WHERE 条件，不要在DataFrame中过滤
分块读取：大数据集使用 chunksize 参数
索引优化：确保查询条件列有数据库索引
类型优化：读取时指定 dtype 使用更紧凑的类型
连接复用：全局复用Engine，避免反复创建连接
批量写入：使用 method='multi' 和 chunksize 参数
避免N+1查询：使用JOIN替代循环查询

8.4 完整工作流示例

"""
完整的数据分析工作流：
从数据库读取 -> 数据清洗 -> 分析计算 -> 结果写回
"""
import pandas as pd
from sqlalchemy import create_engine, text
from sqlalchemy.types import VARCHAR, Float, DateTime
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 1. 建立连接
engine = create_engine(
    'sqlite:///analytics.db',
    echo=False
)

# 2. 从数据库读取原始数据
logger.info("开始读取订单数据...")
df_raw = pd.read_sql_query("""
    SELECT
        o.order_id,
        o.customer_id,
        o.order_date,
        o.status,
        oi.product_id,
        oi.quantity,
        oi.unit_price
    FROM orders o
    JOIN order_items oi ON o.order_id = oi.order_id
    WHERE o.order_date >= '2025-01-01'
      AND o.status IN ('completed', 'shipped')
""", engine, parse_dates=['order_date'])
logger.info(f"读取 {len(df_raw)} 条记录")

# 3. 数据清洗与特征工程
df = df_raw.copy()
df['total_amount'] = df['quantity'] * df['unit_price']
df['month'] = df['order_date'].dt.to_period('M')

# 4. 分析计算
monthly_stats = df.groupby('month').agg(
    order_count=('order_id', 'nunique'),
    total_revenue=('total_amount', 'sum'),
    avg_order_value=('total_amount', 'mean')
).reset_index()

monthly_stats['month'] = monthly_stats['month'].astype('str')

# 5. 结果写回数据库
monthly_stats.to_sql(
    'monthly_analysis',
    engine,
    if_exists='replace',
    index=False,
    dtype={
        'month': VARCHAR(7),
        'order_count': Float(),
        'total_revenue': Float(precision=2),
        'avg_order_value': Float(precision=2)
    }
)

logger.info("分析结果已写入 monthly_analysis 表")

# 6. 验证结果
result = pd.read_sql_table('monthly_analysis', engine)
print(result)

九、核心要点总结

数据库数据读取核心要点：

统一接口：优先使用 pd.read_sql() 通用接口，它能自动区分SQL语句和表名
SQLAlchemy引擎：Engine对象是线程安全的，应在应用生命周期内全局复用
参数化查询：始终使用 params 参数传递变量，防止SQL注入并提升性能
连接管理：使用 with engine.connect() 上下文管理器，自动归还连接到连接池
分块处理：大数据集使用 chunksize 参数分块迭代处理，避免内存溢出
批量写入：使用 method='multi' 结合 chunksize 实现高性能写入
类型映射：理解SQL类型到Pandas类型的映射关系，必要时使用 dtype 和 parse_dates 手动控制
内存优化：使用 float32/int32 替代默认的 float64/int64 可节省50%内存
错误处理：实现重试机制和指数退避策略，提升系统稳定性
if_exists策略：生产环境默认使用 'fail'，全量刷新用 'replace'，增量用 'append'

学习路线建议：先掌握SQLite（零配置，适合练习）→ 再学习MySQL（Web应用最常用）→ 最后深入PostgreSQL（功能最强大，支持JSON/全文搜索等高级特性）。无论使用哪种数据库，通过SQLAlchemy统一接口操作，代码迁移成本极低。