4、数据源接入（二）：数据库数据源（MySQL、PostgreSQL）的连接与增量抽取，使用SQLAlchemy

上一章我们聊了文件数据源，这次轮到数据库了。

说实话，在实际的数据流水线里，数据库才是真正的大头。我经手的项目里，十有八九的数据源都是关系型数据库——MySQL、PostgreSQL 占了绝大多数。你想想看，业务系统、用户订单、财务记录，全在库里躺着。

这一章，我们就来搞定两件事：怎么连上数据库，以及怎么高效地把增量数据抽出来。工具嘛，就用 Python 生态里最成熟的 SQLAlchemy。

4.1 为什么选 SQLAlchemy？

你可能用过 pymysql、psycopg2 这些原生驱动。它们没问题，但有个痛点——换数据库就得改代码。今天用 MySQL，明天切 PostgreSQL，连接参数、SQL 方言全得调一遍。

SQLAlchemy 的核心价值在于抽象层。它把数据库连接、会话管理、ORM 映射都封装好了。你写一套代码，换数据库只需要改连接字符串。我个人习惯用它的 Core 模式（也就是 SQL Expression Language）做数据抽取，轻量、可控，比 ORM 更适合 ETL 场景。

核心优势：
  • 统一的连接接口，切换数据库只需改 dialect
  • 内置连接池，避免频繁创建连接
  • 支持原生 SQL 和表达式两种写法
  • 事务管理完善，出错回滚有保障

4.2 连接 MySQL 和 PostgreSQL

先装依赖。嗯，这里要注意，SQLAlchemy 本身不包含数据库驱动，你得额外装。

# MySQL 用 pymysql 或 mysqlclient
pip install sqlalchemy pymysql

# PostgreSQL 用 psycopg2
pip install sqlalchemy psycopg2-binary

连接字符串的格式长这样：

# MySQL
mysql+pymysql://user:password@host:3306/dbname?charset=utf8mb4

# PostgreSQL
postgresql+psycopg2://user:password@host:5432/dbname

我在项目中遇到过一个问题：密码里包含特殊字符（比如 @、#），直接拼在 URL 里会解析失败。解决办法是用 urllib.parse.quote 编码一下。

from urllib.parse import quote_plus
from sqlalchemy import create_engine

password = quote_plus("your@password#123")
engine = create_engine(f"mysql+pymysql://user:{password}@host:3306/dbname")

创建引擎时，我建议你加上这些参数：

参数 作用 推荐值
pool_size 连接池大小 5-10
max_overflow 超出 pool_size 的最大连接数 10
pool_recycle 连接回收时间（秒） 3600
echo 是否打印 SQL 日志 False（调试时可开） 
engine = create_engine(
    "mysql+pymysql://user:password@host:3306/dbname",
    pool_size=5,
    max_overflow=10,
    pool_recycle=3600,
    echo=False
)
小技巧：连接池的 pool_recycle 一定要设。MySQL 默认 8 小时断开空闲连接，不设回收时间的话，第二天早上跑任务大概率报 "Lost connection" 错误。我吃过这个亏。

4.3 全量抽取 vs 增量抽取

先搞清楚概念。全量抽取就是每次把整张表拉下来，简单粗暴。但数据量一大，这招就不灵了——几百万行每天全量，网络和存储都扛不住。

增量抽取只拿新增或变更的数据。常见的实现方式有三种：

  1. 时间戳增量：根据 updated_at、create_time 字段筛选
  2. 自增 ID 增量：记录上次最大 ID，下次只取更大的
  3. 日志解析（CDC）：解析 binlog、WAL，不依赖业务字段

这一章我们重点讲前两种，CDC 后面会有专门章节。

4.4 时间戳增量抽取实战

这是最常用的方式。前提是表里有个记录变更时间的字段，比如 updated_at。

逻辑很简单：

  • 第一次跑：全量抽取，记录当前最大时间戳
  • 后续跑：只取 updated_at > 上次记录的时间戳 的数据
from sqlalchemy import text
from datetime import datetime

def incremental_extract_by_time(engine, table_name, time_column, last_max_time):
    """
    时间戳增量抽取
    :param engine: SQLAlchemy 引擎
    :param table_name: 表名
    :param time_column: 时间戳字段名
    :param last_max_time: 上次抽取的最大时间
    :return: 本次抽取的数据列表
    """
    query = text(f"""
        SELECT *
        FROM {table_name}
        WHERE {time_column} > :last_time
        ORDER BY {time_column} ASC
    """)
    
    with engine.connect() as conn:
        result = conn.execute(query, {"last_time": last_max_time})
        rows = result.fetchall()
        
        # 更新最大时间戳
        if rows:
            new_max_time = getattr(rows[-1], time_column)
        else:
            new_max_time = last_max_time
            
    return rows, new_max_time
注意：时间戳增量有个坑——如果同一秒内有多条数据写入，且你的查询粒度是秒级，可能会漏数据。我曾经因为这个丢过几百条订单记录。解决办法是把时间戳精度提到毫秒，或者配合 ID 做二次去重。

4.5 自增 ID 增量抽取实战

如果表没有时间字段，但有自增主键，可以用 ID 增量。这种方式更可靠，因为 ID 不会重复、不会回退。

def incremental_extract_by_id(engine, table_name, id_column, last_max_id, batch_size=10000):
    """
    自增 ID 增量抽取（支持分批）
    :param engine: SQLAlchemy 引擎
    :param table_name: 表名
    :param id_column: 自增 ID 字段名
    :param last_max_id: 上次抽取的最大 ID
    :param batch_size: 每批大小
    :return: 生成器，每次返回一批数据
    """
    current_max_id = last_max_id
    
    while True:
        query = text(f"""
            SELECT *
            FROM {table_name}
            WHERE {id_column} > :last_id
            ORDER BY {id_column} ASC
            LIMIT :batch_size
        """)
        
        with engine.connect() as conn:
            result = conn.execute(query, {
                "last_id": current_max_id,
                "batch_size": batch_size
            })
            batch = result.fetchall()
            
            if not batch:
                break
                
            current_max_id = getattr(batch[-1], id_column)
            yield batch
            
            # 如果这批数据少于 batch_size，说明取完了
            if len(batch) < batch_size:
                break

为什么用生成器？你想想看，如果一次抽 100 万行，内存直接爆了。分批返回，每批处理完就释放，稳得很。

4.6 两种增量方式的对比

维度 时间戳增量 自增 ID 增量
依赖字段 需要 updated_at 等时间字段 需要自增主键
数据一致性 可能漏掉同一时间戳的数据 严格递增，不会漏
历史数据修改 能捕获到（updated_at 会变） 无法捕获（ID 不变）
性能 依赖索引，大表需优化 主键索引，性能稳定
适用场景 有明确变更时间的业务表 日志表、流水表（只增不改）

4.7 增量状态管理

每次跑完增量，你得把 last_max_id 或 last_max_time 存下来。不然下次跑不知道从哪开始。

我一般用两种方式：

  • 文件存储：简单场景，存 JSON 或 YAML 文件
  • 数据库表：生产环境，建一张 etl_checkpoint 表
-- checkpoint 表结构
CREATE TABLE etl_checkpoint (
    source_name VARCHAR(100) PRIMARY KEY,
    last_max_id BIGINT,
    last_max_time DATETIME,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
def save_checkpoint(engine, source_name, last_max_id=None, last_max_time=None):
    query = text("""
        INSERT INTO etl_checkpoint (source_name, last_max_id, last_max_time, updated_at)
        VALUES (:name, :max_id, :max_time, NOW())
        ON DUPLICATE KEY UPDATE
            last_max_id = VALUES(last_max_id),
            last_max_time = VALUES(last_max_time),
            updated_at = NOW()
    """)
    with engine.connect() as conn:
        conn.execute(query, {
            "name": source_name,
            "max_id": last_max_id,
            "max_time": last_max_time
        })
        conn.commit()
建议：checkpoint 表和生产数据表放在同一个数据库里，这样事务可以保证一致性。如果 checkpoint 存 Redis，万一 Redis 挂了，增量可能重复跑。

4.8 本章知识体系

下面这张图把本章的核心逻辑串起来了。从连接数据库开始，到选择增量策略，再到状态持久化，一条线走到底。

数据库增量抽取流程 数据库增量抽取核心流程 MySQL / PostgreSQL SQLAlchemy Engine 选择增量策略 时间戳增量 自增ID增量 Checkpoint 持久化 记录 last_max_id 或 last_max_time

说白了，增量抽取的核心就三步：连上库、选对策略、记住位置。策略选错了，数据可能重复或遗漏；位置没记住，下次全量重跑，等于白干。

我个人更倾向于自增 ID 增量，因为它简单、可靠、不容易出幺蛾子。但现实中的表不一定有自增 ID，所以时间戳增量也得掌握。两种都学会，遇到什么表都不慌。

← 上一章 📖 返回目录 下一章 →