做爬虫最烦的一件事是什么?不是被限制,而是第二天跑脚本发现昨天抓过的数据又抓了一遍。尤其当你维护的是一个长期更新的数据源,每次全量抓取,既浪费时间又浪费带宽,还可能给对方服务器造成不必要的压力。增量采集这个概念,就是专门解决这个问题的——只抓新增或变化的数据,跳过已经抓过的部分。这一节我把 last_time 和 last_id 这两种最主流的增量游标策略,从原理到实操各做一遍,你跟着敲完,就能直接拿去改自己手上的项目。
这套内容适合已经会写最基础的 Python 爬虫(会用 requests 和 XPath 或解析库),但还没接触过“如何让爬虫可持续运行”的读者。增量采集不是某个库,而是一套设计思想:用一个可持久化的标记,记录“上次抓到哪里”,下次从这个标记继续。理解了这一点,市面上绝大多数列表型数据源的增量抓取你都能自己搭出来。
1. 增量采集到底解决什么问题:先搞清楚你的爬虫在重复做什么
很多初学者默认“爬虫就是一次性把页面全抓下来”,但这只是学习阶段的错觉。真正投入使用的爬虫,面对的是持续更新的数据:新闻网站每分钟发稿,电商平台每小时上架,论坛帖子实时在变。如果每次运行都从第一页开始抓到最后一页,你会遇到三个非常具体的浪费。
1.1 全量抓取的三个典型浪费场景
第一个浪费是重复请求。假设目标网站有 500 页列表数据,你昨天抓完了全部,今天再全量跑一遍,其中 490 页的内容都没变,等于白白多发了 490 次请求。这不仅是时长问题,还会让对方的访问日志里出现大量高频请求,轻则影响对方服务,重则让运维把你的 IP 拉进黑名单。
第二个浪费是数据清洗和去重成本。全量抓回来以后,你仍然需要和数据库里的旧数据比对,判断哪些是新插入、哪些是更新、哪些是删除。这个逻辑写起来并不简单,而且数据量越大,去重越慢,最后你会发现大部分时间不是在爬,而是在做无意义的比对。
第三个浪费是异常定位困难。全量抓取时,如果任务运行到一半挂了,你很难判断“哪些已经抓完、哪些还没开始”。因为没有一个明确的进度标记,下次只能从头再来。而增量采集天然自带“断点续传”属性——游标记录到哪里,就从哪里继续。
1.2 增量采集的定义与适用前提
增量采集,就是每次只抓取自上次采集以来新增或发生变更的数据。它依赖一个核心概念:游标(cursor),也就是记录采集进度的标记。游标可以是时间戳(last_time),也可以是自增 ID(last_id),还可以是页码、偏移量等其他形式,但那一对最经典的就是前两者。
不过增量采集不是什么场景都能套用,它有一个非常明确的前提:目标数据源必须存在某种单调变化的字段。所谓单调变化,就是这个字段的值只朝一个方向走,比如发布时间的值只会越来越新,自增 ID 只会越来越大。没有这个前提,你就没有可靠的下一次起点。
另外还要注意,增量采集解决的是“新增”问题,不解决“修改”问题。如果你的数据源允许旧记录被编辑(比如帖子被更新、价格被调整),那么光靠时间游标和 ID 游标是不够的,还需要加上“按更新时间全量扫描最近一段窗口”这类混合策略。这一节先用最简单的增量模型把原理打通,混合策略以后可以在这个基础上扩展。
1.3 两条路线 last_time / last_id 的本质区别
last_time 和 last_id 看起来都是“记一个数”,但它们的适用对象完全不同。
last_time 记录的是时间,代表“我只抓取某个时间点之后的数据”。它的优势是语义直观,和目标网站上的“发布时间”“更新时间”字段一一对应。劣势是时间有精度上限,如果同一秒内生成了多条数据,而且你又只记到秒,就可能漏掉一部分。
last_id 记录的是数字 ID,代表“我只抓取某个 ID 之后的数据”。它的优势是严格递增时永不重复、永不遗漏,而且不需要依赖服务器返回的日期字段。劣势是必须确认目标数据的 ID 真的是递增的,有些网站的 ID 是随机字符串甚至雪花算法生成的,时间有序但无法直接比较大小。
这两个策略不是谁取代谁,而是互补。下面两节我会各写一遍完整实现,你就能体会到它们各自在工程里的真实手感。
2. 策略一实战:基于 last_time 的时间游标增量采集
先说 last_time。这是大多数人第一次接触增量采集时最先想到的做法,因为它跟你用 Excel 筛选“今天新增”的习惯一模一样。核心逻辑就一句话:把上次抓到的最后一条时间记下来,下次请求参数里的时间起点从它开始。
2.1 last_time 策略的原理与适用场景
last_time 适合列表型页面里带明确时间字段的数据,比如新闻列表、博客文章、聊天记录、操作日志。这些页面通常支持按时间倒序排列,而且列表接口支持传入起始时间参数。
它的工作流程是这样的:第一次运行时,游标是空的或一个很早的时间(比如 1970-01-01),爬虫从第一页开始抓,抓完以后把当前批次里最大(最新)的时间更新为新的 last_time 并保存。第二次运行时,读取 last_time,把它作为请求起始参数,爬虫只请求这个时间之后的内容。每抓完一页,再用这一页里的最新时间放回游标,直到遇到某页里所有数据的时间都不超过 last_time,就说明新增数据已经取完,任务可以结束。
这里最关键的细节是:游标更新的时机要选对。我见过不少人在开始抓之前就把 last_time 设成当前时间,结果运行中发布的新数据全部漏掉。正确做法应该是每翻一页就动态使用当前页的最新时间,因为目标数据源可能在你抓的过程中不断产生新数据,如果你锁死起点,就会漏掉“采集过程中才出现的增量”。
2.2 手写一个带时间游标的爬虫核心代码
假设目标是一个新闻列表接口,返回 JSON,里面每条数据有id、title、publish_time三个字段,并且接口支持start_time参数,返回结果按时间倒序排列。我们用 requests 加一个最简单的 JSON 文件来持久化游标。
import json import time import requests CURSOR_FILE = "cursor_time.json" BASE_URL = "https://example.com/api/news" def load_cursor(): try: with open(CURSOR_FILE, "r", encoding="utf-8") as f: data = json.load(f) return data.get("last_time", "1970-01-01 00:00:00") except FileNotFoundError: return "1970-01-01 00:00:00" def save_cursor(last_time): with open(CURSOR_FILE, "w", encoding="utf-8") as f: json.dump({"last_time": last_time}, f, ensure_ascii=False, indent=2) def fetch_news(start_time, page=1): params = { "start_time": start_time, "page": page, "page_size": 50, } resp = requests.get(BASE_URL, params=params, timeout=10) resp.raise_for_status() return resp.json() def collect_new_items(): cursor = load_cursor() page = 1 latest_time = cursor while True: data = fetch_news(cursor, page=page) items = data.get("items", []) if not items: break # 当前页所有数据都早于或等于上一次的游标,说明没有新增了 newest_on_page = items[0]["publish_time"] if newest_on_page <= cursor: break for item in items: # 同样要过滤掉等于游标的旧数据,避免重复入库 if item["publish_time"] > cursor: # 这里写你的入库逻辑 print(f"新增: {item['id']} - {item['title']} - {item['publish_time']}") if item["publish_time"] > latest_time: latest_time = item["publish_time"] # 翻页到最后一页就结束 if page >= data.get("total_pages", 1): break page += 1 # 每处理一页就更新一次游标,防止任务中断后丢失进度 save_cursor(latest_time) time.sleep(1) save_cursor(latest_time) if __name__ == "__main__": collect_new_items()注意这里有个“双保险”逻辑:一边用cursor作为请求参数来控制接口返回范围,一边在入库前用item["publish_time"] > cursor做二次过滤。接口不一定完全按时间排序,或者同一批次里混进了少量旧数据,这一步可以把它们挡在外面。
2.3 时间格式、时区与边界条件的处理细节
凡是牵扯到时间,就必然牵扯到格式和时区。第一个建议是:跟接口请求和比较都用字符串。只要目标接口返回的时间格式是稳定的(比如都是YYYY-MM-DD HH:MM:SS),字符串比较天然就是时间比较,因为这种格式的字典序等于时间顺序。不要轻易把字符串转成 datetime 再转回来,转一次就可能引入时区偏差。
第二个建议是:游标初始值不要用空字符串。第一次运行时要给它一个“足够早”的时间,1970-01-01 00:00:00是常见选择。但如果你发现目标网站上还有更早的数据,就改用0000-00-00或者目标系统支持的最小时间,保证第一轮能全量抓完。
第三个建议是:时间精度问题要克制。我看到很多教程喜欢用毫秒甚至微秒来存 last_time,这在抓取高并发生成的数据时确实有帮助,但也带来了字符串解析复杂度。我的习惯是一开始先看目标接口的时间精度,如果对方只给到秒,我游标就用秒;如果对方给到毫秒,我再用毫秒。跟随源数据精度,是最省事的做法。
提示:last_time 策略最怕“同一秒内写入大量数据”。假设你一页能抓到 50 条,而某一秒内发布了 500 条,翻页过程中因为按秒过滤,很可能把一部分数据划到游标之外。遇到这种情况,要么改用 ID 策略,要么把时间窗口重叠处理——比如游标回拨 10 秒,再在入库前去重。
3. 策略二实战:基于 last_id 的数字游标增量采集
如果你发现目标页面没有可靠的时间字段,或者时间参数控制不了接口返回范围,那就应该立刻想到 last_id。它不依赖任何日期,只需要列表数据里有一个递增的 ID。
3.1 last_id 策略的工作原理与适用场景
last_id 是电商商品、论坛帖子、视频列表这类场景里的主力方案。这类数据的 ID 通常是数据库自增主键,新记录一定比旧记录 ID 大。我们用游标记住上次最大的 ID,下一次请求只拉取大于这个 ID 的数据。
它比 last_time 更干净的地方在于:没有时区问题,没有精度问题,没有“同一秒并发导致漏数据”的烦恼。只要 ID 严格递增,游标就能做到既不重复也不遗漏。
但它的适用条件也很苛刻:ID 必须能代表插入顺序。有些网站对外展示的 ID 是混淆过的(比如随机哈希),有些是雪花算法生成的(时间上有序但数值上不方便比较),还有些是 UUID。遇到这些情况,last_id 就不成立了,需要回到 last_time 或者改用下一页游标。实操前,先花五分钟拉几条数据看一眼 ID 增长的规律,这个时间花得很值。
3.2 手写一个带 ID 游标的爬虫核心代码
还是假设一个 JSON 接口,这次不传时间参数,改成传offset_id或min_id,返回按 ID 倒序。
import json import time import requests CURSOR_FILE = "cursor_id.json" BASE_URL = "https://example.com/api/items" def load_cursor(): try: with open(CURSOR_FILE, "r", encoding="utf-8") as f: data = json.load(f) return int(data.get("last_id", 0)) except (FileNotFoundError, ValueError): return 0 def save_cursor(last_id): with open(CURSOR_FILE, "w", encoding="utf-8") as f: json.dump({"last_id": last_id}, f, ensure_ascii=False, indent=2) def fetch_items(min_id, page=1): params = { "min_id": min_id, "page": page, "page_size": 50, } resp = requests.get(BASE_URL, params=params, timeout=10) resp.raise_for_status() return resp.json() def collect_new_items(): cursor = load_cursor() page = 1 max_id = cursor while True: data = fetch_items(cursor, page=page) items = data.get("items", []) if not items: break # 如果是按 ID 倒序返回,第一项的 ID 就是当前批次最大值 if items[0]["id"] <= cursor: break for item in items: # 二次过滤,护栏永不省略 if item["id"] > cursor: print(f"新增: {item['id']} - {item['title']}") if item["id"] > max_id: max_id = item["id"] if page >= data.get("total_pages", 1): break page += 1 save_cursor(max_id) time.sleep(1) save_cursor(max_id) if __name__ == "__main__": collect_new_items()逻辑跟 last_time 几乎一一对应,只是把时间比较换成了数字比较。要注意的是初始值0是否合适。如果这张表里的 ID 是从 1 开始的,那没问题;如果历史数据里有 ID 为负数的极端情况,就把初始游标调成-1或者取第一轮最小 ID 减一。
3.3 ID 乱序、删除与追加场景的处理
last_id 有个经典误区:以为接口返回的 ID 一定是从大到小排列。但很多列表接口受排序规则影响(比如按热度排、按随机排),返回顺序和 ID 大小没有关系。这时候如果你还用“第一项 ID 作为游标”,就会导致游标乱跳。
正确做法是:遍历当前页所有数据,自己维护一个max_id,取整个批次里的最大 ID。同时,判断“是否还有下一页”也不能依赖列表顺序,而要看接口是否返回了has_more之类的标记,或者继续向后翻页直到某页所有 ID 都不大于游标。
另一个常见情况是数据删除。假设上次游标是 5000,下次运行时目标数据源的 5001 到 5020 已经被删了,直接返回 5021。你可能会奇怪:为什么新增数据只有 5021 之后的那几条?其实这是对的,删除操作不影响增量采集的正确性,你只需要把 5021 之后的新数据入库即可。真正要小心的是反向删除:如果数据源允许物理删除 ID 较大的记录,可能会造成游标回退,但这种情况在绝大多数对外接口里不会发生,因为删除通常只是假删除,不影响 ID 生成。
最后是“ID 自增但有空洞”的情况。这个非常常见,比如事务回滚、数据预分配都会造成 ID 不连续。你完全不用管,增量采集只关心“大于游标的 ID”,哪怕中间缺了 100 个号,你也只需要抓真正存在的那些。
4. 两种策略的踩坑经验对比:我三类踩坑场景的完整处理记录
学完上面两节,你已经会写两种游标了。但真实项目里踩坑的往往不是核心逻辑,而是边界和小细节。这一节我把两种策略放在一起做一个全景对比,顺便复盘我自己在项目里真正踩过的坑。
4.1 last_time 与 last_id 的核心参数对比
| 对比维度 | last_time | last_id |
|---|---|---|
| 适用数据 | 带明确时间字段的列表 | 带递增 ID 的列表 |
| 游标内容 | 最近一次采集到的最新时间 | 最近一次采集到的最大 ID |
| 请求参数 | 通常传 start_time 或 begin_time | 通常传 min_id 或 offset_id |
| 重复风险 | 时间精度不足时可能重复 | 极低,ID 严格递增时不重复 |
| 遗漏风险 | 同一秒内大量新增可能遗漏 | 极低,遇上 ID 乱序才可能有风险 |
| 时区影响 | 有,必须统一时区 | 无 |
| 数据删除影响 | 不影响 | 基本不影响 |
| 修改旧数据影响 | 无法感知 | 无法感知 |
| 适合阶段 | 新闻、日志、交易记录 | 商品、帖子、视频列表 |
这个表格不是让你背的,是在选型时做快速判断用的。我的经验是:能拿到稳定 ID 就优先用 last_id;只有确实只有时间字段可用时,才退回到 last_time。last_id 的实现更简单,心智负担更小。
4.2 我踩过的三个真实坑
第一个坑:游标保存太频繁导致文件写得越来越慢。早期我在每个for循环里保存一次游标,数据量一旦上千,JSON 文件的写入频率就变成了性能瓶颈。后来改成“每处理完一页保存一次”,速度立刻起来了。你会问,那中途断了会不会丢进度?会丢当前页的进度,代价很小,重新跑一次即可,不会重复入库前一条已入库的数据,因为有入库前过滤兜底。
第二个坑:时间字符串比较翻车。有个目标接口返回的时间格式是2024/03/01 12:00:00,而我程序里存的游标老格式是2024-03-01 12:00:00。字符串比较时按下划线顺序,/的 ASCII 码(47)比-的 ASCII 码(45)大,导致时间比较完全错乱。解决方案很简单:在入库前统一做一个格式归一化,把所有来源的时间都转成同一种字符串格式。
第三个坑:ID 总被误判为“时间字段”。有次我偷懒,发现列表里带一个雪花 ID,以为可以用 last_id。结果那个 ID 不是单调递增的,而是在时间上大致递增但数值上存在跳跃和倒挂。抓了几轮以后才发现漏了一大批数据。后来我学乖了,选型之前一定要先拉 100 条数据看 ID 的变化规律,不要只看样本里的前几条。
4.3 游标持久化:从 JSON 文件到 SQLite 的工程化升级
游标存什么地方,看项目阶段。
单机小项目、学习项目,用 JSON 文件就够了,代码少、直观、好调试。但 JSON 文件有一个问题:多个爬虫任务并发运行时,后写的会覆盖先写的,游标互相污染。这时候就要升级到 SQLite或者数据库表。我一般用一个极其简单的表结构来管理:
CREATE TABLE crawl_cursor ( task_name TEXT PRIMARY KEY, cursor_value TEXT, updated_at TEXT );这样每个任务都有自己的游标记录,互不干扰,而且查历史、回滚游标都很方便。配合 SQLite 的INSERT OR REPLACE或者UPDATE,十几行代码就能封装好。
import sqlite3 class CursorStore: def __init__(self, db_path="cursor.db"): self.conn = sqlite3.connect(db_path) self.conn.execute("""CREATE TABLE IF NOT EXISTS crawl_cursor ( task_name TEXT PRIMARY KEY, cursor_value TEXT, updated_at TEXT )""") def get(self, task_name): cur = self.conn.execute( "SELECT cursor_value FROM crawl_cursor WHERE task_name = ?", (task_name,) ) row = cur.fetchone() return row[0] if row else None def set(self, task_name, cursor_value): self.conn.execute( "INSERT OR REPLACE INTO crawl_cursor (task_name, cursor_value, updated_at) VALUES (?, ?, ?)", (task_name, cursor_value, __import__("time").strftime("%Y-%m-%d %H:%M:%S")), ) self.conn.commit()这段代码把游标管理从业务逻辑里剥出来了,之后不管你的爬虫是跑在定时任务里,还是跑在 Web 后端里,都能复用同一个存储方案。
5. 把增量采集封装成可复用模块,并顺手解决正确性验证
前面两节的代码是为了讲清原理,结构其实是“平铺”的。真实项目里我不会把游标读写散落在每个函数里,而是会封装成一个模块。这里给你看一下我常用的封装思路,以及怎么验证你的增量采集到底有没有写对。
5.1 用 CursorStore 统一管理游标,让爬虫代码瘦身
把游标读写统一之后,业务代码可以收敛成下面这种结构:
from cursor_store import CursorStore store = CursorStore("crawl.db") last_id = int(store.get("item_task") or 0) items = fetch_items(min_id=last_id) new_items = [i for i in items if i["id"] > last_id] save_to_database(new_items) new_max = max(i["id"] for i in new_items) store.set("item_task", str(new_max))这样做的好处是,你的爬虫函数里不再出现打开文件、关闭文件、解析 JSON 之类的杂活。以后如果要从 SQLite 换成 Redis,只需要改CursorStore这一个类,业务代码一行都不用动。
5.2 增量采集的正确性验证:重复跑一遍是最快的测试
增量采集到底有没有搞对,有一个很粗暴又有效的验证方法:连续跑两遍,看第二遍有没有新增数据。正常的增量采集,第二遍应该输出 0 条新增。如果第二遍居然还有数据输出,说明你的游标没更新,或者边界过滤写错了。
我常用的验证清单是这样的:
- 第一次全量跑完,记录当前入库总数记为 A。
- 什么都不改,第二次再跑,入库总数应该还是 A。
- 手动往数据表里插一条模拟新数据(或等真实数据更新),第三次跑,总数应该变成 A+1。
- 查看游标文件或游标表,确认游标值等于最近一次采集到的最大 ID 或最新时间。
- 手动把游标回退一步,再跑一次,确认旧数据不会重复入库(因为代码里有
>过滤)。
这套流程跑通之后,你才能放心把爬虫交给定时任务。
5.3 让增量采集持续推进:定时任务与日志观察
最后是调度。增量采集大多跑在服务器上,我习惯用系统 cron 来触发,避免额外引入一堆依赖。假设你每天 8 点跑一次:
0 8 * * * cd /path/to/project && /usr/bin/python3 run_incremental.py >> log.log 2>&1日志很重要。不要只打印“开始”“结束”,要打印每次采集的游标变化,格式大概是:
INFO 2025-01-10 08:00:01 task=item_task old_cursor=15231 new_cursor=15388 added=157有了这种日志,一旦增量采集漏数据,翻日志就能很快定位是哪一天、哪一个游标开始出错的。没有日志的爬虫等于裸奔,这话我说过很多次。
我个人在实际项目里还有一个习惯:增量任务跑到后半段时,会顺手把游标值打印出来。看到游标在稳定增长,心理就有底了。如果你也打算长期维护一个数据源,强烈建议你从这节开始,就把游标管理和业务抓取彻底拆开——这让后续加规则、换数据源、接监控都轻松很多。