Scrapy招聘职位爬虫：字段设计、清洗去重与增量采集实战-平芜编程栈

简介：这是一份围绕 Scrapy 框架爬取智联招聘职位信息的小型实战项目，面向已掌握 Python 基础、希望理解爬虫工程化流程的初学者与进阶学习者，可用于练习职位名称、公司、薪资、地点等字段的抓取与整理。压缩包为 zip 格式，共 10 个文件，以 7 个 py 源码文件为主，辅以 2 个 pyc 编译文件和 1 个 cfg 配置，整体约 6KB，体量轻便，适合快速导入本地环境阅读和运行。项目按爬虫、数据项、管道、中间件、配置等模块组织，涉及请求调度、页面解析、登录验证、反爬处理与结果持久化等关键环节，能帮助读者理清 Scrapy 各组件的协作方式，并理解从规则定义到数据落地的完整链路，目录层级清晰，便于按模块阅读与二次修改，也方便在现有骨架上扩展字段和存储方式。已有 536 人学习，可作为课程作业、爬虫入门练习或小型采集脚本的参考样例。

1. Scrapy 爬招聘职位：先把“能跑”与“能维护”分开

用 Scrapy 爬取招聘职位信息，最容易翻车的不是选择器写错，而是字段口径没定、翻页断在第二页、重跑一遍数据全重复。这套 scrapypython 项目骨架把列表页解析、详情页跟进、字段清洗、入库去重拆成四段，目标不是抓一次就完，而是下次换关键词或换城市还能接着跑。它适合需要做岗位趋势、薪资分布统计的 Python 开发者，也适合刚学 Scrapy 想找一个完整链路练手的人；边界要先说清楚：只处理公开列表页和详情页，不碰登录后简历数据，不做验证码绕过，频率要压到站点能接受的范围。拿到代码后先跑通一页，再谈增量。

2. 项目骨架与字段设计：从列表页到详情页的两次解析

2.1 为什么先定 Item 再写 Spider

很多人一上来就写 parse，边写边 print，字段名随缘。跑到 Pipeline 才发现 salary 是字符串、city 里带区、发布时间有“今天”“昨天”，后面清洗和入库全得返工。Item 是 Spider 和 Pipeline 之间的契约，先定字段，后面清洗、导出、建表才有落点。我一般先跑 50 条样本，缺失率超过 20% 的字段先不加，避免字段表看起来很全、实际全是空列。

import scrapy class JobItem(scrapy.Item): job_title = scrapy.Field() # 职位名称，保留原始文本 company = scrapy.Field() # 招聘方名称，只做展示 salary_raw = scrapy.Field() # 原始薪资文本，如 15-25K·13薪 city = scrapy.Field() # 城市 district = scrapy.Field() # 区县，可为空 experience_raw = scrapy.Field() # 原始经验要求 education_raw = scrapy.Field() # 原始学历要求 publish_time = scrapy.Field() # 发布时间文本或标准化日期 job_url = scrapy.Field() # 规范化后的详情页 URL job_desc = scrapy.Field() # 职位描述纯文本 list_page = scrapy.Field() # 来源列表页页码，排查用 crawl_time = scrapy.Field() # 采集时间，ISO 格式 fingerprint = scrapy.Field() # URL 指纹，入库去重用

Field 本身不限制类型，但建议在 Pipeline 里转成明确类型。带_raw的字段保留原文，标准化字段另存，清洗规则改坏时还能回看原始证据。job_url要规范化，去掉追踪参数；list_page是排查翻页问题的线索；crawl_time用datetime.utcnow().isoformat()写 ISO 格式，后面做增量时好比较。

2.2 列表页翻页与详情页跟进

先建项目骨架，命令不要手敲错：

scrapy startproject jobspider cd jobspider scrapy genspider jobs example.com

项目名用jobspider，Spider 名用jobs，allowed_domains先用示例域名占位，实际换掉。下面这段 Spider 同时处理列表页翻页和详情页跟进：

import scrapy from jobspider.items import JobItem class JobsSpider(scrapy.Spider): name = "jobs" allowed_domains = ["example.com"] start_urls = ["https://example.com/jobs?page=1"] max_pages = 20 def parse(self, response): page = response.meta.get("page", 1) for card in response.css(".job-list-item"): href = card.css(".job-title a::attr(href)").get() if not href: continue yield response.follow( href, callback=self.parse_detail, meta={"list_page": page}, ) next_href = response.css(".pager .next::attr(href)").get() if next_href and page < self.max_pages: yield response.follow( next_href, callback=self.parse, meta={"page": page + 1}, ) def parse_detail(self, response): item = JobItem() item["job_title"] = response.css("h1::text").get(default="").strip() item["company"] = response.css(".company-name::text").get(default="").strip() item["salary_raw"] = response.css(".salary::text").get(default="").strip() item["city"] = response.css(".location a::text").get(default="").strip() item["experience_raw"] = response.css(".experience::text").get(default="").strip() item["education_raw"] = response.css(".education::text").get(default="").strip() item["publish_time"] = response.css(".publish-time::text").get(default="").strip() item["job_desc"] = " ".join(response.css(".job-detail ::text").getall()).strip() item["job_url"] = response.url item["list_page"] = response.meta.get("list_page") yield item

parse里先遍历列表卡片，拿详情页链接后交给parse_detail，同时判断下一页链接是否还存在。response.follow会自动处理相对 URL，不要自己拼字符串。max_pages是保险丝，防止下一页选择器失效后无限翻页；meta把list_page带到详情页，出问题时能知道是哪一页的数据。

选择器是占位写法，实际页面结构要先在 shell 里验证：

scrapy shell "https://example.com/jobs?page=1"

进入 shell 后看len(response.css(".job-list-item"))是否大于 0，再看.job-title a::attr(href)能不能取到链接。不要一边猜选择器一边跑全量，翻页和详情页两个解析函数混在一起时，先确认列表页能出链接，再确认详情页能出字段。

2.3 用 Feed Export 先落一份可校验的 JSONL

在接数据库之前，先把数据落到 JSONL。它比 CSV 更能保留嵌套文本，比直接入库更容易回滚：

# settings.py FEEDS = { "jobs.jsonl": { "format": "jsonlines", "encoding": "utf8", "overwrite": True, "fields": [ "job_title", "company", "salary_raw", "city", "district", "experience_raw", "education_raw", "publish_time", "job_url", "list_page", "crawl_time", "fingerprint", ], } } FEED_EXPORT_ENCODING = "utf-8" LOG_LEVEL = "INFO"

format用jsonlines，一行一条，边跑边能tail。overwrite为 True 时每次覆盖，避免追加导致重复；fields固定导出顺序，方便 diff。先限制页数跑一遍：

scrapy crawl jobs -s CLOSESPIDER_PAGECOUNT=25

CLOSESPIDER_PAGECOUNT是页数上限，不是条数上限；如果要按条数停，用CLOSESPIDER_ITEMCOUNT。跑完做两项校验：

wc -l jobs.jsonl head -n 1 jobs.jsonl | python -m json.tool python - <<'PY' import json from collections import Counter missing = Counter() with open("jobs.jsonl", encoding="utf-8") as f: for line in f: row = json.loads(line) for key in ("job_title", "job_url", "city", "salary_raw"): if not row.get(key): missing[key] += 1 print(missing) PY

先看行数是否和预期量级接近，再看首行 JSON 是否合法，最后统计核心字段空值。JSONL 只是中间产物，不要直接当最终库；它的价值是让你在接 MySQL 之前，先把字段覆盖率、重复率和解析错误暴露出来。

3. 采集节流与异常处理：并发、延迟、重试的参数怎么定

3.1 并发与延迟的取舍：先算目标 QPS

并发和延迟不是拍脑袋填的。先定目标：每分钟大概请求多少次，再反推参数。粗略公式是QPS ≈ CONCURRENT_REQUESTS / (DOWNLOAD_DELAY + 平均响应时间)。如果平均响应 0.4 秒，DOWNLOAD_DELAY=1.0，全局并发 8，理论 QPS 大约 5.7，但单域名并发还会再压一层。招聘列表页和详情页通常不需要高并发，先把单站点并发压在 2 到 4 之间。

# settings.py CONCURRENT_REQUESTS = 8 CONCURRENT_REQUESTS_PER_DOMAIN = 4 DOWNLOAD_DELAY = 1.0 RANDOMIZE_DOWNLOAD_DELAY = True AUTOTHROTTLE_ENABLED = True AUTOTHROTTLE_START_DELAY = 1.0 AUTOTHROTTLE_MAX_DELAY = 10.0 AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0

CONCURRENT_REQUESTS是全局并发，CONCURRENT_REQUESTS_PER_DOMAIN才是对单个站点生效的并发。DOWNLOAD_DELAY是同一域名两次请求之间的间隔，RANDOMIZE_DOWNLOAD_DELAY会在 0.5 到 1.5 倍之间随机。AUTOTHROTTLE会根据响应延迟自动拉长等待，TARGET_CONCURRENCY不要设太高，2 到 4 更适合公开列表页。先跑 50 页，看日志里有没有 429 或 503，再决定要不要继续降。

注意：只采集公开页面，控制请求频率，不碰登录后才能看到的数据。频率设置不是技术炫技，是让任务能持续跑完。

3.2 重试与超时：哪些异常该重试，哪些该丢弃

重试不是越多越好。403 重试不会变成 200，404 重试也不会变出页面，429 要的是退避而不是立刻重发。比较稳的配置是把 5xx、408、429 放进重试列表，把 403 和 404 排除：

RETRY_ENABLED = True RETRY_TIMES = 2 RETRY_HTTP_CODES = [500, 502, 503, 504, 522, 524, 408, 429] DOWNLOAD_TIMEOUT = 15

RETRY_TIMES=2是额外重试次数，不是总请求次数。DOWNLOAD_TIMEOUT对列表页 15 秒够用，详情页字段多时可以放宽到 20 秒。再给 Spider 加一个errback，把失败请求记下来，而不是把失败请求重新 yield 成新 Request，否则重试计数会失真：

from scrapy.spidermiddlewares.httperror import HttpError from twisted.internet.error import DNSLookupError, TimeoutError def errback(self, failure): if failure.check(HttpError): response = failure.value.response self.logger.warning( "http_error status=%s url=%s", response.status, response.url, ) elif failure.check(TimeoutError, DNSLookupError): self.logger.warning("network_error url=%s", failure.request.url)

日志里重点看retry/count、httperror/forbidden、downloader/response_status_count/429。如果重试数在涨、Item 数不涨，说明请求在被拒绝，继续加并发只会更糟。先降CONCURRENT_REQUESTS_PER_DOMAIN，再把DOWNLOAD_DELAY拉到 2 秒，观察一轮。

3.3 请求头与 Cookie 的边界：只做合规的会话维持

请求头不需要伪装成十几种浏览器。固定一个项目标识的 User-Agent，配上和目标页面一致的语言头，反而更容易排查问题：

DEFAULT_REQUEST_HEADERS = { "Accept": "text/html,application/xhtml+xml", "Accept-Language": "zh-CN,zh;q=0.9", } USER_AGENT = "JobSpider/1.0 (+https://example.com/bot)" COOKIES_ENABLED = True

USER_AGENT写项目名和说明地址，不要用随机 UA 列表硬撞。Accept-Language跟目标页面语言一致，能减少返回移动版或空壳页的概率。COOKIES_ENABLED默认开启，用来维持普通会话；不要塞登录 Cookie 去抓权限外页面，这是边界问题，不是技术问题。

如果换机器后第一页就 403，先检查response.text是不是验证页，再看请求头是否少了Accept或Referer，最后才怀疑频率。不要一上来就加并发或换请求头组合，那会把问题搅得更难定位。

4. 数据清洗与入库：Pipeline 里的去重、补全和幂等

4.1 薪资、地点、经验字段的归一化

原始字段直接入库，后面做统计会很难受。薪资、地点、经验这三类字段先写纯函数，别把正则散落在 Spider 里：

import re def parse_salary(text: str): if not text or "面议" in text: return None, None, None, None m = re.search(r"(\d+(?:\.\d+)?)-(\d+(?:\.\d+)?)\s*([Kk千万元]?)", text) if not m: return None, None, None, None low, high = float(m.group(1)), float(m.group(2)) unit = m.group(3).upper() or "K" months = 12 mm = re.search(r"(\d+)\s*薪", text) if mm: months = int(mm.group(1)) return low, high, unit, months def split_location(text: str): if not text: return "", "" parts = re.split(r"[·\-\s]", text, maxsplit=1) return parts[0], parts[1] if len(parts) > 1 else "" def parse_experience(text: str): if not text: return None, None if "应届" in text or "无经验" in text: return 0, 1 m = re.search(r"(\d+)-(\d+)年", text) if m: return int(m.group(1)), int(m.group(2)) m = re.search(r"(\d+)年以上", text) if m: return int(m.group(1)), None return None, None

薪资只做辅助分析，不把“13薪”直接乘进月薪，而是单独存salary_months。面议返回None而不是 0，避免均值被拉低。地点按第一个分隔符切，城市和区县分开；经验“3-5年”映射成上下限，“5年以上”上限留空。正则要按实际文本扩充，比如“元/天”“万/月”这些格式，先拿样本跑一遍再定稿。

4.2 用 URL 指纹做增量去重

Scrapy 默认的RFPDupeFilter只对请求去重，数据层重复它管不了。同一个职位可能从列表页出现两次，也可能带不同追踪参数。先规范化 URL，再算指纹：

import hashlib from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit TRACKING_KEYS = {"from", "source", "utm_source", "utm_medium", "utm_campaign"} def normalize_url(url: str) -> str: parts = urlsplit(url) query = [(k, v) for k, v in parse_qsl(parts.query) if k not in TRACKING_KEYS] path = parts.path.rstrip("/") or "/" return urlunsplit( (parts.scheme, parts.netloc.lower(), path, urlencode(query), "") ) def fingerprint(url: str) -> str: return hashlib.sha1(normalize_url(url).encode("utf-8")).hexdigest()

去追踪参数、域名小写、路径去

本文还有配套的精品资源，点击获取