做爬虫最烦的一件事是什么不是被限制而是第二天跑脚本发现昨天抓过的数据又抓了一遍。尤其当你维护的是一个长期更新的数据源每次全量抓取既浪费时间又浪费带宽还可能给对方服务器造成不必要的压力。增量采集这个概念就是专门解决这个问题的——只抓新增或变化的数据跳过已经抓过的部分。这一节我把 last_time 和 last_id 这两种最主流的增量游标策略从原理到实操各做一遍你跟着敲完就能直接拿去改自己手上的项目。这套内容适合已经会写最基础的 Python 爬虫会用 requests 和 XPath 或解析库但还没接触过“如何让爬虫可持续运行”的读者。增量采集不是某个库而是一套设计思想用一个可持久化的标记记录“上次抓到哪里”下次从这个标记继续。理解了这一点市面上绝大多数列表型数据源的增量抓取你都能自己搭出来。1. 增量采集到底解决什么问题先搞清楚你的爬虫在重复做什么很多初学者默认“爬虫就是一次性把页面全抓下来”但这只是学习阶段的错觉。真正投入使用的爬虫面对的是持续更新的数据新闻网站每分钟发稿电商平台每小时上架论坛帖子实时在变。如果每次运行都从第一页开始抓到最后一页你会遇到三个非常具体的浪费。1.1 全量抓取的三个典型浪费场景第一个浪费是重复请求。假设目标网站有 500 页列表数据你昨天抓完了全部今天再全量跑一遍其中 490 页的内容都没变等于白白多发了 490 次请求。这不仅是时长问题还会让对方的访问日志里出现大量高频请求轻则影响对方服务重则让运维把你的 IP 拉进黑名单。第二个浪费是数据清洗和去重成本。全量抓回来以后你仍然需要和数据库里的旧数据比对判断哪些是新插入、哪些是更新、哪些是删除。这个逻辑写起来并不简单而且数据量越大去重越慢最后你会发现大部分时间不是在爬而是在做无意义的比对。第三个浪费是异常定位困难。全量抓取时如果任务运行到一半挂了你很难判断“哪些已经抓完、哪些还没开始”。因为没有一个明确的进度标记下次只能从头再来。而增量采集天然自带“断点续传”属性——游标记录到哪里就从哪里继续。1.2 增量采集的定义与适用前提增量采集就是每次只抓取自上次采集以来新增或发生变更的数据。它依赖一个核心概念游标cursor也就是记录采集进度的标记。游标可以是时间戳last_time也可以是自增 IDlast_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, encodingutf-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, encodingutf-8) as f: json.dump({last_time: last_time}, f, ensure_asciiFalse, indent2) def fetch_news(start_time, page1): params { start_time: start_time, page: page, page_size: 50, } resp requests.get(BASE_URL, paramsparams, timeout10) 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, pagepage) 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, encodingutf-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, encodingutf-8) as f: json.dump({last_id: last_id}, f, ensure_asciiFalse, indent2) def fetch_items(min_id, page1): params { min_id: min_id, page: page, page_size: 50, } resp requests.get(BASE_URL, paramsparams, timeout10) 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, pagepage) 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_timelast_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_pathcursor.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_idlast_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。手动往数据表里插一条模拟新数据或等真实数据更新第三次跑总数应该变成 A1。查看游标文件或游标表确认游标值等于最近一次采集到的最大 ID 或最新时间。手动把游标回退一步再跑一次确认旧数据不会重复入库因为代码里有过滤。这套流程跑通之后你才能放心把爬虫交给定时任务。5.3 让增量采集持续推进定时任务与日志观察最后是调度。增量采集大多跑在服务器上我习惯用系统 cron 来触发避免额外引入一堆依赖。假设你每天 8 点跑一次0 8 * * * cd /path/to/project /usr/bin/python3 run_incremental.py log.log 21日志很重要。不要只打印“开始”“结束”要打印每次采集的游标变化格式大概是INFO 2025-01-10 08:00:01 taskitem_task old_cursor15231 new_cursor15388 added157有了这种日志一旦增量采集漏数据翻日志就能很快定位是哪一天、哪一个游标开始出错的。没有日志的爬虫等于裸奔这话我说过很多次。我个人在实际项目里还有一个习惯增量任务跑到后半段时会顺手把游标值打印出来。看到游标在稳定增长心理就有底了。如果你也打算长期维护一个数据源强烈建议你从这节开始就把游标管理和业务抓取彻底拆开——这让后续加规则、换数据源、接监控都轻松很多。