Python定时从SQL Server导出数据时如何避免重复获取？

下面提供几种无需修改目标数据表的实现思路，按推荐优先级排序：

1. 基于时间戳字段（优先推荐，若表存在创建/更新时间列） #

如果你的数据表本身带有create_time或update_time这类记录数据生成时间的字段，可以通过记录上次运行时的最新时间，每次查询仅获取该时间之后的新增数据。

实现步骤： #

• 用本地文件存储上次获取到的最大时间戳
• 首次运行时获取全量数据，并记录数据中的最大时间戳
• 后续运行时，仅查询时间戳大于上次记录值的数据
• 更新本地存储的时间戳为本次获取到的最大时间戳

代码示例： #

import pyodbc
import datetime
import os

# 存储上次运行时间戳的文件
LAST_TIMESTAMP_FILE = "last_fetch_timestamp.txt"

def load_last_timestamp():
    if os.path.exists(LAST_TIMESTAMP_FILE):
        with open(LAST_TIMESTAMP_FILE, "r") as f:
            return datetime.datetime.fromisoformat(f.read().strip())
    return None

def save_last_timestamp(timestamp):
    with open(LAST_TIMESTAMP_FILE, "w") as f:
        f.write(timestamp.isoformat())

def fetch_incremental_data():
    # 建立SQL Server连接
    conn_str = "DRIVER={SQL Server};SERVER=你的服务器地址;DATABASE=你的数据库;UID=用户名;PWD=密码"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    last_ts = load_last_timestamp()
    current_max_ts = None

    if last_ts is None:
        # 首次运行，拉取全量数据
        print("首次运行，获取全量数据")
        cursor.execute("SELECT * FROM 你的表名")
    else:
        # 增量拉取，筛选时间戳大于上次记录的数据
        print(f"增量获取自 {last_ts} 之后的新数据")
        cursor.execute("SELECT * FROM 你的表名 WHERE create_time > ?", (last_ts,))

    # 获取并处理数据
    rows = cursor.fetchall()
    for row in rows:
        # 这里替换成你的数据处理逻辑（如写入文件、分析等）
        print(row)

    # 更新本次获取的最大时间戳
    cursor.execute("SELECT MAX(create_time) FROM 你的表名")
    current_max_ts = cursor.fetchone()[0]
    if current_max_ts:
        save_last_timestamp(current_max_ts)
    else:
        # 表为空时，用当前时间作为初始标记
        save_last_timestamp(datetime.datetime.now())

    conn.close()

if __name__ == "__main__":
    fetch_incremental_data()

2. 基于自增主键/唯一递增标识 #

如果数据表有自增ID（如id INT IDENTITY(1,1)）或其他严格递增的唯一字段，可以通过记录上次获取到的最大ID值，后续仅查询ID大于该值的数据。

代码示例： #

import pyodbc
import os

# 存储上次获取的最大ID的文件
LAST_ID_FILE = "last_fetch_max_id.txt"

def load_last_max_id():
    if os.path.exists(LAST_ID_FILE):
        with open(LAST_ID_FILE, "r") as f:
            return int(f.read().strip())
    return None

def save_last_max_id(max_id):
    with open(LAST_ID_FILE, "w") as f:
        f.write(str(max_id))

def fetch_incremental_data():
    conn_str = "DRIVER={SQL Server};SERVER=你的服务器地址;DATABASE=你的数据库;UID=用户名;PWD=密码"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    last_max_id = load_last_max_id()
    current_max_id = None

    if last_max_id is None:
        print("首次运行，获取全量数据")
        cursor.execute("SELECT * FROM 你的表名")
    else:
        print(f"增量获取ID大于 {last_max_id} 的新数据")
        cursor.execute("SELECT * FROM 你的表名 WHERE id > ?", (last_max_id,))

    rows = cursor.fetchall()
    for row in rows:
        print(row)

    # 更新本次获取的最大ID
    cursor.execute("SELECT MAX(id) FROM 你的表名")
    current_max_id = cursor.fetchone()[0]
    if current_max_id:
        save_last_max_id(current_max_id)

    conn.close()

if __name__ == "__main__":
    fetch_incremental_data()

3. 无时间戳/自增ID时的兜底方案（小表适用） #

如果数据表既没有时间戳也没有递增标识，只能通过记录已获取过的记录的唯一特征（如多字段组合的唯一键），每次拉取全量数据后对比筛选新增记录。注意：此方法效率较低，仅适合数据量较小的表。

代码示例： #

import pyodbc
import os
import pickle

# 存储已获取记录唯一标识的文件
FETCHED_RECORDS_FILE = "fetched_records.pkl"

def load_fetched_records():
    if os.path.exists(FETCHED_RECORDS_FILE):
        with open(FETCHED_RECORDS_FILE, "rb") as f:
            return pickle.load(f)
    return set()

def save_fetched_records(records_set):
    with open(FETCHED_RECORDS_FILE, "wb") as f:
        pickle.dump(records_set, f)

def fetch_incremental_data():
    conn_str = "DRIVER={SQL Server};SERVER=你的服务器地址;DATABASE=你的数据库;UID=用户名;PWD=密码"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    fetched_records = load_fetched_records()
    current_records = set()
    new_rows = []

    # 拉取全量数据
    cursor.execute("SELECT col1, col2, col3 FROM 你的表名")
    all_rows = cursor.fetchall()

    # 生成唯一标识并筛选新增记录（这里用col1+col2作为唯一键，根据实际表结构调整）
    for row in all_rows:
        record_key = f"{row[0]}_{row[1]}"
        current_records.add(record_key)
        if record_key not in fetched_records:
            new_rows.append(row)

    if not fetched_records:
        print("首次运行，获取全量数据")
    else:
        print(f"增量获取 {len(new_rows)} 条新数据")

    for row in new_rows:
        print(row)

    # 更新已获取记录的标识集合
    save_fetched_records(current_records)

    conn.close()

if __name__ == "__main__":
    fetch_incremental_data()

注意事项： #

• 本地存储的标记文件（如时间戳、ID文件）需确保权限正常，避免被误删除
• 时间戳方案需注意SQL Server与Python运行环境的时区一致性，避免漏取或重复获取数据
• 若脚本运行间隔较长，需确保期间没有数据被删除（如果需要处理删除的记录，此方案不适用）

内容的提问来源于stack exchange，提问作者Sai ganesh Dokala