一、前言招聘平台的岗位数据是求职分析、行业薪资调研、企业人才画像的核心数据源但手动采集效率极低且不同平台BOSS 直聘 / 智联 / 51job的技术架构差异大BOSS 直聘 / 51job 采用动态接口渲染传统页面解析无法获取完整数据智联招聘需结合地理编码补充经纬度信息且分页逻辑易出现死循环多平台数据格式不统一易出现重复、字段缺失、薪资格式混乱等问题。本文基于Scrapy框架搭建一站式招聘数据爬取系统适配三大平台的爬取逻辑解决「反爬封禁、数据重复、格式不规范、存储杂乱」四大核心问题最终实现结构化、无重复、可直接分析的招聘数据采集与存储。二、技术栈选型技术 / 工具用途选型原因Scrapy爬虫核心框架异步高并发成熟的爬虫生态支持 Pipeline、中间件等扩展DrissionPage浏览器自动化 接口监听无需分析复杂加密接口直接监听前端请求获取原始 JSON 数据规避反爬Redis分布式去重SETNX 原子操作实现岗位 ID 唯一判定支持多爬虫进程 / 机器协同去重MySQL数据持久化存储维度建模公司 / 地区 / 技能维度表 岗位事实表满足结构化分析需求高德地图 API地理编码地址转经纬度补充智联招聘缺失的经纬度字段支持后续地理维度分析正则表达式薪资格式标准化解析兼容 K / 万 / 千 / 日薪等多格式薪资转为数字类型便于分析三、项目整体架构整个系统遵循「数据采集→标准化→去重→维度存储」的流程核心架构如下四、项目核心实现4.1 项目结构搭建首先创建标准的 Scrapy 项目并规划目录结构# 1. 创建Scrapy项目 scrapy startproject boss cd boss # 2. 规划核心目录 mkdir -p boss/spiders # 存放三大平台爬虫 touch boss/items.py # 标准化Item定义 touch boss/job_dict.py # 岗位大类配置 touch boss/pipelines.py# 去重存储Pipeline touch boss/settings.py # 全局配置 touch run.py # 爬虫启动脚本最终项目结构boss/ ├── boss/ │ ├── __init__.py │ ├── items.py # 标准化字段定义 │ ├── job_dict.py # 岗位大类映射配置 │ ├── pipelines.py # Redis去重MySQL存储 │ ├── settings.py # Scrapy全局配置 │ └── spiders/ │ ├── __init__.py │ ├── 51job.py # 51job爬虫接口监听 │ ├── zp.py # 智联爬虫页面解析高德API │ └── zhipin.py # BOSS直聘爬虫接口监听 └── run.py # 统一启动脚本4.2 基础模块定义4.2.1 标准化 Item 定义items.py统一三大平台的字段格式避免数据混乱import scrapy class BossItem(scrapy.Item): # 核心标识字段 job_id scrapy.Field() # 岗位唯一ID去重核心 # 岗位基础信息 岗位名称 scrapy.Field() 标准岗位 scrapy.Field() # 标准化岗位名如“大数据开发” 岗位大类 scrapy.Field() # 岗位所属大类如“数据开发” # 公司信息 公司 scrapy.Field() 公司领域 scrapy.Field() 规模 scrapy.Field() # 岗位要求 薪资 scrapy.Field() 学历要求 scrapy.Field() 经验要求 scrapy.Field() 技能需求 scrapy.Field() # 地理信息 市 scrapy.Field() 区 scrapy.Field() 商圈 scrapy.Field() 经度 scrapy.Field() # 数字类型 纬度 scrapy.Field() # 数字类型 # 扩展信息 搜索关键词 scrapy.Field() 来源渠道 scrapy.Field() # 如“BOSS直聘/智联/51job” 访问地址 scrapy.Field() # 岗位详情URL4.2.2 岗位大类配置job_dict.py定义岗位标准化映射规则统一多平台的岗位分类JOB_BIG_TYPE { 数据开发: [ 数据开发工程师, 数仓工程师, Hive工程师, Spark工程师, ETL工程师, 大数据工程师 ], 数据分析: [ 数据分析师, 商业分析师, BI分析师, 数据运营, 数据产品分析 ], 后端开发: [ Java开发工程师, Python后端, Go开发工程师, 后台开发 ], 算法: [ 算法工程师, 机器学习工程师, 推荐算法, 搜索算法, NLP工程师 ], 产品: [ 产品经理, 数据产品经理, 商业产品经理 ] } def normalize_job(job_name): 标准化岗位名称匹配大类和标准岗位 for big_type, jobs in JOB_BIG_TYPE.items(): for j in jobs: if j in job_name: return big_type, j return 其他, 其他 def build_search_keywords(): 构建去重后的搜索关键词列表 keywords [] for jobs in JOB_BIG_TYPE.values(): keywords.extend(jobs) return list(set(keywords))4.3 三大平台爬虫实现4.3.1 51job 爬虫接口监听模式51job 采用动态接口加载数据使用DrissionPage监听前端接口请求直接获取原始 JSON 数据避免页面解析的繁琐import scrapy import time import json from DrissionPage import ChromiumPage from ..job_dict import JOB_BIG_TYPE, normalize_job, build_search_keywords from ..items import BossItem class Job51Spider(scrapy.Spider): name 51job # 禁用无关中间件降低反爬风险 custom_settings { SPIDER_MIDDLEWARES: { scrapy.spidermiddlewares.offsite.OffsiteMiddleware: None, scrapy.spidermiddlewares.referer.RefererMiddleware: None, } } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 初始化浏览器开启接口监听 self.dp ChromiumPage() self.dp.listen.start(search-pc) # 监听51job搜索接口 self.logger.info(浏览器初始化完成接口监听已开启) def start_requests(self): # 启动请求触发爬虫逻辑 yield scrapy.Request( urlhttps://we.51job.com, callbackself.parse_with_drission, dont_filterTrue ) def parse_with_drission(self, response): # 遍历所有搜索关键词 for keyword in build_search_keywords(): self.logger.info(f采集关键词{keyword}) # 构造搜索URL上海040000可替换为其他城市 search_url fhttps://we.51job.com/pc/search?jobArea040000keyword{keyword} self.dp.get(search_url) # 获取第一页数据 resp self.dp.listen.wait(timeout10) first_data resp.response.body if isinstance(first_data, str): first_data json.loads(first_data) # 提取总页数容错处理 total_page first_data.get(resultbody, {}).get(job, {}).get(totalcount, 1) try: total_page max(int(total_page), 1) except (ValueError, TypeError): total_page 1 self.logger.info(f{keyword} 共 {total_page} 页数据) # 解析第一页岗位 job_list first_data.get(resultbody, {}).get(job, {}).get(items, []) if not job_list: self.logger.info(f{keyword} 无数据) continue yield from self.parse_jobs(job_list, keyword) # 翻页爬取最多20页避免反爬 max_page min(total_page, 20) for page in range(2, max_page 1): try: self.dp.scroll.to_bottom() next_btn self.dp.ele(css:button.btn-next, timeout5) if not next_btn: self.logger.info(f{keyword} 已无下一页) break next_btn.click() self.logger.info(f正在采集{keyword} 第{page}页的数据!) resp self.dp.listen.wait(timeout10) data resp.response.body if isinstance(data, str): data json.loads(data) job_list data.get(resultbody, {}).get(job, {}).get(items, []) if not job_list: break yield from self.parse_jobs(job_list, keyword) time.sleep(1.5) # 延时反爬 except Exception as e: self.logger.warning(f{keyword} 第 {page} 页异常: {e}) break def parse_jobs(self, job_list, keyword): 解析岗位数据生成标准化Item for job in job_list: try: item BossItem() job_name job.get(jobName, ) big_type, std_job normalize_job(job_name) area job.get(jobAreaLevelDetail, {}) # 填充标准化字段 item[job_id] job.get(jobId) item[岗位名称] job_name item[标准岗位] std_job item[岗位大类] big_type item[公司] job.get(fullCompanyName) item[公司领域] job.get(industryType1Str) item[规模] job.get(companySizeString) item[薪资] job.get(provideSalaryString) item[学历要求] job.get(degreeString) item[经验要求] job.get(workYearString) # 技能标签跳过前2个无关标签 job_tags_slice job.get(jobTags, [])[2:] item[技能需求] ,.join(job_tags_slice) if job_tags_slice else 暂无 item[市] area.get(cityString) item[区] area.get(districtString) item[商圈] job.get(landmarkString) item[经度] job.get(lon) item[纬度] job.get(lat) item[搜索关键词] keyword item[来源渠道] 51job item[访问地址] job.get(jobHref) yield item except Exception as e: self.logger.warning(f单个岗位解析失败跳过{str(e)}) continue time.sleep(5) def closed(self, reason): 爬虫关闭时释放浏览器资源 self.logger.info(f爬虫开始关闭原因{reason}) self.dp.listen.stop() self.dp.quit() self.logger.info(浏览器已关闭资源释放完成)4.3.2 智联招聘爬虫页面解析 高德地理编码智联招聘需解析静态页面且无原生经纬度字段集成高德地图 API 补充地理编码同时用 Redis 做 URL 去重import scrapy from scrapy import Request, signals import redis import requests import re from ..items import BossItem class ZpSpider(scrapy.Spider): name zp allowed_domains [www.zhaopin.com] start_urls [https://www.zhaopin.com/sou/jl765/kwB4JMAS33DPFG0KUH/p1] max_page 50 # 限制最大翻页数 classmethod def from_crawler(cls, crawler, *args, **kwargs): s cls() s._set_crawler(crawler) crawler.signals.connect(s.spider_opened, signalsignals.spider_opened) crawler.signals.connect(s.spider_closed, signalsignals.spider_closed) return s def spider_opened(self, spider): # 初始化Redis清空旧数据 self.red redis.Redis(hostlocalhost, port6379, db1, decode_responsesTrue) self.redis_key zhilian:job_urls self.red.delete(self.redis_key) self.logger.info(Redis旧数据已清空开始全新抓取) def parse(self, response): # 提取当前页岗位详情URL data_list response.xpath(//div[classjobinfo__top]/a/href).getall() self.logger.info(f当前页{response.url}提取到{len(data_list)}个岗位URL) for url in data_list: href response.urljoin(url) if self.red.sismember(self.redis_key, href): self.logger.debug(f岗位URL已访问跳过{href}) continue self.logger.info(f发起岗位详情请求{href}) self.red.sadd(self.redis_key, href) yield Request(href, callbackself.parse_job_detail, meta{url: href}) # 翻页逻辑仅抓取“下一页”避免页码混乱 next_page_url response.xpath( //div[classsoupager]/a[contains(text(), 下一页)]/href ).get(default) if not next_page_url: next_page_url response.xpath(//div[classsoupager]/a[last()]/href).get(default) if next_page_url: next_page_href response.urljoin(next_page_url) current_page int(response.url.split(/p)[-1]) if p in response.url else 1 next_page int(next_page_href.split(/p)[-1]) if p in next_page_href else 1 if next_page current_page and next_page self.max_page and not self.red.sismember(self.redis_key, next_page_href): self.logger.info(f发起下一页请求第{next_page}页URL{next_page_href}) self.red.sadd(self.redis_key, next_page_href) yield Request(next_page_href, callbackself.parse, meta{url: next_page_href}) else: self.logger.info(f已达到最大分页数或下一页无效停止分页) def parse_job_detail(self, response): 解析岗位详情页补充高德地理编码 item BossItem() url response.url # 提取岗位ID job_id re.search(rjobdetail/([A-Z0-9])\.htm, url).group(1) if re.search(rjobdetail/([A-Z0-9])\.htm, url) else item[job_id] job_id # 核心字段提取 item[岗位名称] response.xpath(//h1[classsummary-plane__title]/text()).get(default).strip() job_name item[岗位名称] # 标准化岗位分类 if 大数据 in job_name: item[标准岗位] 大数据开发 item[岗位大类] 数据开发 elif 后台 in job_name or 后端 in job_name: item[标准岗位] 后台开发 item[岗位大类] 后端开发 else: item[标准岗位] item[岗位大类] item[公司] response.xpath(//a[classcompany__title]/text()).get(default).strip() # 公司领域去重 company_industry response.xpath(//button[classcompany__industry]/text()).get(default).strip() if company_industry: industry_list list(dict.fromkeys([i.strip() for i in company_industry.split(,)])) item[公司领域] ,.join(industry_list) else: item[公司领域] item[规模] response.xpath(//button[classcompany__size and not(contains(., 融资))]/text()).get(default).strip() item[薪资] response.xpath(//span[classsummary-plane__salary]/text()).get(default).strip() # 学历/经验提取 info_list response.xpath(//ul[classsummary-plane__info]/li/text()).getall() item[学历要求] item[经验要求] for info in info_list: info_clean info.strip() if info_clean in [本科, 大专, 硕士, 博士]: item[学历要求] info_clean elif 年 in info_clean or 应届 in info_clean: item[经验要求] info_clean # 技能需求 skills_list response.xpath(//span[classdescribtion__skills-item]/text()).getall() item[技能需求] [skill.strip() for skill in skills_list if skill.strip()] # 地址信息 city response.xpath(//ul[classsummary-plane__info]/li[1]/a/text()).get(default).strip() district response.xpath(//ul[classsummary-plane__info]/li[1]/span/text()).get(default).strip() business_circle response.xpath(//span[classjob-address__content-text]/text()).get(default).strip() item[市] city item[区] district # 清洗商圈文本 business_circle re.sub(r^\s*\[图标\]\s*|\s*$, , business_circle).replace(district, ).strip() item[商圈] business_circle # 高德地理编码获取经纬度替换为你的API Key amap_api_key 你的高德API Key # 免费申请https://lbs.amap.com/ full_address if business_circle: full_address f{city}{district}{business_circle} elif district: full_address f{city}{district} elif city: full_address city item[经度] 0.0 item[纬度] 0.0 if full_address and amap_api_key: try: api_url ( fhttps://restapi.amap.com/v3/geocode/geo f?address{requests.utils.quote(full_address)} foutputjson fkey{amap_api_key} ) resp requests.get(api_url, timeout10) resp.raise_for_status() api_data resp.json() if api_data.get(status) 1 and len(api_data.get(geocodes, [])) 0: location api_data[geocodes][0].get(location, ) if location: longitude, latitude location.split(,) item[经度] float(longitude) item[纬度] float(latitude) self.logger.debug(f地址{full_address} 经纬度获取成功{longitude},{latitude}) except Exception as e: self.logger.warning(f经纬度获取失败地址{full_address}{str(e)}) # 固定字段 item[搜索关键词] 大数据开发 item[来源渠道] 智联 item[访问地址] url.strip() or 暂无 # Redis去重 href response.meta.get(url) if href and not self.red.sismember(self.redis_key, href): self.red.sadd(self.redis_key, href) self.logger.debug(f岗位数据抓取完成{item[岗位名称]}) yield item def spider_closed(self, spider): self.red.save() self.red.close() self.logger.info(爬虫正常关闭Redis连接已关闭)4.3.3 BOSS 直聘爬虫接口监听模式与 51job 逻辑类似适配 BOSS 直聘的接口监听规则核心差异在于接口标识和字段映射import scrapy import time from DrissionPage import ChromiumPage from ..job_dict import JOB_BIG_TYPE, normalize_job, build_search_keywords from ..items import BossItem class ZhipinSpider(scrapy.Spider): name zhipin custom_settings { SPIDER_MIDDLEWARES: { scrapy.spidermiddlewares.offsite.OffsiteMiddleware: None, scrapy.spidermiddlewares.referer.RefererMiddleware: None, } } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.dp ChromiumPage() self.dp.listen.start(joblist) # BOSS直聘岗位列表接口标识 self.logger.info(浏览器初始化完成接口监听已开启) def start_requests(self): yield scrapy.Request( urlhttps://www.zhipin.com, callbackself.parse_with_drission, dont_filterTrue ) def parse_with_drission(self, response): for keyword in build_search_keywords(): self.logger.info(f采集关键词{keyword}) # 深圳101280600可替换为其他城市 url fhttps://www.zhipin.com/web/geek/jobs?query{keyword}city101280600 self.dp.get(url) # 爬取前20页 for page in range(1,21): self.logger.info(f正在采集关键词【{keyword}】第{page}页数据!) resp self.dp.listen.wait() data resp.response.body or {} job_list data.get(zpData, {}).get(jobList, []) for job in job_list: try: item BossItem() job_name job.get(jobName, ) big_type, std_job normalize_job(job_name) gps job.get(gps) or {} item[job_id] job.get(encryptJobId) item[岗位名称] job.get(jobName) item[标准岗位] std_job item[岗位大类] big_type item[公司] job.get(brandName) item[公司领域] job.get(brandIndustry) item[规模] job.get(brandScaleName) item[薪资] job.get(salaryDesc) item[学历要求] job.get(jobDegree) item[经验要求] job.get(jobExperience) item[技能需求] job.get(skills) item[市] job.get(cityName) item[区] job.get(areaDistrict) item[商圈] job.get(businessDistrict) item[经度] gps.get(longitude) item[纬度] gps.get(latitude) item[搜索关键词] keyword item[来源渠道] BOSS直聘 item[访问地址] fhttps://www.zhipin.com/job_detail/{job.get(encryptJobId, )}.html if job.get(encryptJobId) else 暂无 yield item except Exception as e: self.logger.warning(f单个岗位解析失败跳过{str(e)}) continue self.dp.scroll.to_bottom() time.sleep(2) self.dp.quit() def closed(self, reason): self.logger.info(f爬虫开始关闭原因{reason}) self.dp.listen.stop() self.dp.quit() self.logger.info(浏览器已关闭资源释放完成)4.4 数据处理 Pipeline4.4.1 Redis 去重 Pipeline基于岗位唯一 ID 实现分布式去重避免重复爬取和存储import redis import scrapy class RedisDedupPipeline: def open_spider(self, spider): self.redis redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue ) def process_item(self, item, spider): # 兼容多平台的岗位ID字段 job_id item.get(job_id) or item.get(encrypt_job_id) or item.get(encryptJobId) if not job_id: raise scrapy.exceptions.DropItem(无 jobId非法岗位) redis_key fboss:job:{job_id} # SETNX仅当key不存在时写入有效期7天 is_new_job self.redis.set(redis_key, 1, ex60*60*24*7, nxTrue) if not is_new_job: raise scrapy.exceptions.DropItem(fRedis 判定为重复岗位job_id{job_id}) return item def close_spider(self, spider): self.redis.connection_pool.disconnect() spider.logger.info(Redis连接已关闭)4.4.2 MySQL 存储 Pipeline采用数仓维度建模思想将数据拆分为「维度表 事实表 桥表」同时实现多格式薪资解析import pymysql import scrapy import datetime import re class MySQLPipeline: def __init__(self): self.conn None self.cursor None def open_spider(self, spider): # 初始化MySQL连接 self.conn pymysql.connect( hostlocalhost, userroot, password123456, # 替换为你的密码 databaseboss_job, # 提前创建数据库 charsetutf8mb4, autocommitTrue ) self.cursor self.conn.cursor() spider.logger.info(MySQL连接已建立) def parse_salary(self, salary_desc): 解析多格式薪资转为数字类型 if not salary_desc or str(salary_desc).strip().lower() in [null, none, nan, ]: return None, None salary_desc str(salary_desc).strip() # 定义薪资格式正则 patterns [ (r(\d(?:\.\d)?)\s*-\s*(\d(?:\.\d)?)\s*K, 1000, None, None), # 15-20K (r(\d(?:\.\d)?)\s*-\s*(\d(?:\.\d)?)\s*万/年, 10000, 12, None), # 20-25万/年 (r(\d(?:\.\d)?)\s*-\s*(\d(?:\.\d)?)\s*万, 10000, None, None), # 1.5-2万 (r(\d(?:\.\d)?)\s*-\s*(\d(?:\.\d)?)\s*千, 1000, None, None), # 6-8千 (r(\d(?:\.\d)?)\s*元/天, 1, None, daily), # 200元/天 ] # 匹配并解析 for pattern, multiplier, divisor, salary_type in patterns: match re.search(pattern, salary_desc, re.IGNORECASE) if match: if salary_type daily: # 日薪转月薪21.75个工作日 daily_salary float(match.group(1)) * multiplier monthly daily_salary * 21.75 return int(monthly), int(monthly) else: min_val float(match.group(1)) * multiplier max_val float(match.group(2)) * multiplier if len(match.groups()) 1 else min_val if divisor: min_val / divisor max_val / divisor return int(min_val), int(max_val) # 处理混合单位如8千-1.6万 if - in salary_desc and (千 in salary_desc or 万 in salary_desc or K in salary_desc.upper()): parts salary_desc.split(-) if len(parts) 2: min_salary self._parse_single_value(parts[0].strip()) max_salary self._parse_single_value(parts[1].strip()) if min_salary and max_salary: return min_salary, max_salary return None, None def _parse_single_value(self, value_str): 解析单个薪资值 if 万 in value_str: match re.search(r(\d(?:\.\d)?), value_str) return int(float(match.group(1)) * 10000) if match else None elif 千 in value_str: match re.search(r(\d(?:\.\d)?), value_str) return int(float(match.group(1)) * 1000) if match else None elif K in value_str.upper(): match re.search(r(\d(?:\.\d)?), value_str) return int(float(match.group(1)) * 1000) if match else None match re.search(r(\d(?:\.\d)?), value_str) return int(float(match.group(1))) if match else None def get_company_id(self, name, industry, scale): 获取/插入公司维度表 name name or 未知公司 industry industry or 未知领域 scale scale or 未知规模 sql INSERT INTO dim_company (company_name, industry, scale) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE company_idLAST_INSERT_ID(company_id) self.cursor.execute(sql, (name, industry, scale)) return self.cursor.lastrowid def get_region_id(self, city, district, business): 获取/插入地区维度表 city city or 未知城市 district district or 未知区域 business business or 未知商圈 sql INSERT INTO dim_region (city, district, business_area) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE region_idLAST_INSERT_ID(region_id) self.cursor.execute(sql, (city, district, business)) return self.cursor.lastrowid def get_skill_id(self, skill): 获取/插入技能维度表 skill skill or 未知技能 sql INSERT INTO dim_skill (skill_name) VALUES (%s) ON DUPLICATE KEY UPDATE skill_idLAST_INSERT_ID(skill_id) self.cursor.execute(sql, (skill,)) return self.cursor.lastrowid def process_item(self, item, spider): job_id item.get(job_id) or item.get(encrypt_job_id) or item.get(encryptJobId) if not job_id: spider.logger.warning(fitem 缺少 job_id跳过写入 MySQL{item}) return item # 插入维度表 company_id self.get_company_id(item.get(公司), item.get(公司领域), item.get(规模)) region_id self.get_region_id(item.get(市), item.get(区), item.get(商圈)) # 解析薪资 salary_min, salary_max self.parse_salary(item.get(薪资)) # 插入岗位事实表 job_sql INSERT INTO fact_job ( job_id, job_name, std_job, job_category, company_id, region_id, salary_desc, salary_min, salary_max, experience, degree, lng, lat, search_keyword, crawl_time,source,job_url ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) job_params ( job_id, item.get(岗位名称) or 未知岗位, item.get(标准岗位) or 未知标准岗位, item.get(岗位大类) or 未知岗位大类, company_id, region_id, item.get(薪资) or 未知薪资, salary_min, salary_max, item.get(经验要求) or 未知经验要求, item.get(学历要求) or 未知学历要求, float(item.get(经度) or 0.0), float(item.get(纬度) or 0.0), item.get(搜索关键词) or 未知关键词, datetime.datetime.now(), item.get(来源渠道) or 未知, item.get(访问地址) or 未知 ) try: self.cursor.execute(job_sql, job_params) spider.logger.info(f成功写入岗位数据到 MySQLjob_id{job_id}) except Exception as e: spider.logger.error(f写入岗位数据失败job_id{job_id}错误{str(e)}) self.conn.rollback() return item # 插入岗位-技能桥表 skills item.get(技能需求, ) or [] if isinstance(skills, str): skills skills.split(,) if skills.strip() else [] for skill in skills: skill skill.strip() if not skill: continue try: skill_id self.get_skill_id(skill) bridge_sql INSERT IGNORE INTO fact_job_skill (job_id, skill_id) VALUES (%s, %s) self.cursor.execute(bridge_sql, (job_id, skill_id)) except Exception as e: spider.logger.error(f写入技能桥表失败job_id{job_id}skill{skill}错误{str(e)}) self.conn.rollback() continue return item def close_spider(self, spider): self.cursor.close() self.conn.close() spider.logger.info(MySQL连接已关闭)4.5 全局配置settings.pyBOT_NAME boss SPIDER_MODULES [boss.spiders] NEWSPIDER_MODULE boss.spiders # 反爬优化 USER_AGENT Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36 ROBOTSTXT_OBEY False LOG_LEVEL INFO DOWNLOAD_DELAY 2 # 请求延时避免反爬 FEED_EXPORT_ENCODING utf-8 # 启用Pipeline先去重后存储 ITEM_PIPELINES { boss.pipelines.RedisDedupPipeline: 100, boss.pipelines.MySQLPipeline: 200, } # 兼容Scrapy 2.x REQUEST_FINGERPRINTER_IMPLEMENTATION 2.7 TWISTED_REACTOR twisted.internet.asyncioreactor.AsyncioSelectorReactor4.6 启动脚本run.pyfrom scrapy.cmdline import execute import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) if __name__ __main__: # 选择要爬取的平台注释/取消注释 # 智联 # execute(scrapy crawl zp.split()) # BOSS直聘 # execute(scrapy crawl zhipin.split()) # 51job execute(scrapy crawl 51job.split())五、环境准备与运行5.1 环境依赖安装pip install scrapy DrissionPage redis pymysql requests5.2 MySQL 表结构创建提前创建数据库和表结构实现维度建模CREATE DATABASE IF NOT EXISTS boss_job DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE boss_job; -- 公司维度表 CREATE TABLE IF NOT EXISTS dim_company ( company_id INT PRIMARY KEY AUTO_INCREMENT, company_name VARCHAR(255) NOT NULL, industry VARCHAR(100), scale VARCHAR(50), UNIQUE KEY uk_company (company_name, industry, scale) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 地区维度表 CREATE TABLE IF NOT EXISTS dim_region ( region_id INT PRIMARY KEY AUTO_INCREMENT, city VARCHAR(50), district VARCHAR(50), business_area VARCHAR(50), UNIQUE KEY uk_region (city, district, business_area) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 技能维度表 CREATE TABLE IF NOT EXISTS dim_skill ( skill_id INT PRIMARY KEY AUTO_INCREMENT, skill_name VARCHAR(50) NOT NULL, UNIQUE KEY uk_skill (skill_name) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 岗位事实表 CREATE TABLE IF NOT EXISTS fact_job ( job_id VARCHAR(100) PRIMARY KEY, job_name VARCHAR(255) NOT NULL, std_job VARCHAR(50), job_category VARCHAR(50), company_id INT, region_id INT, salary_desc VARCHAR(50), salary_min INT, salary_max INT, experience VARCHAR(50), degree VARCHAR(20), lng FLOAT, lat FLOAT, search_keyword VARCHAR(50), crawl_time DATETIME, source VARCHAR(20), job_url VARCHAR(500), FOREIGN KEY (company_id) REFERENCES dim_company(company_id), FOREIGN KEY (region_id) REFERENCES dim_region(region_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 岗位-技能桥表 CREATE TABLE IF NOT EXISTS fact_job_skill ( job_id VARCHAR(100), skill_id INT, PRIMARY KEY (job_id, skill_id), FOREIGN KEY (job_id) REFERENCES fact_job(job_id), FOREIGN KEY (skill_id) REFERENCES dim_skill(skill_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;5.3 启动与验证5.3.1 启动爬虫# 启动51job爬虫 python run.py # 或直接通过Scrapy命令启动 scrapy crawl zhipin # BOSS直聘 scrapy crawl zp # 智联5.3.2 数据验证Redis 验证查看去重的岗位 IDredis-cli keys boss:job:* # 查看所有岗位IDMySQL 验证查询结构化数据-- 查询岗位公司薪资信息 SELECT f.job_name, f.salary_min, f.salary_max, d.company_name, d.industry FROM fact_job f LEFT JOIN dim_company d ON f.company_id d.company_id LIMIT 10;六、避坑指南与优化6.1 核心避坑点反爬规避切勿删除DOWNLOAD_DELAY建议设置≥2 秒单平台单次爬取≤20 页避免账号 / IP 被封禁智联招聘需替换自己的高德 API Key免费额度足够个人使用字段容错所有字段均做get容错避免单个字段缺失导致爬虫崩溃资源释放爬虫均实现closed钩子确保浏览器 / 数据库连接正常释放薪资解析覆盖 K / 万 / 千 / 日薪等多格式避免薪资数据无法量化分析。6.2 进阶优化方向分布式爬取集成Scrapy-Redis实现多机器分布式爬取提升效率定时爬取结合APScheduler实现每日定时爬取更新岗位数据可视化分析基于Pandas/Matplotlib/Plotly实现薪资分布、技能需求等可视化异常告警添加邮件 / 钉钉告警爬取失败时及时通知数据清洗增加字段标准化清洗如公司规模统一格式、学历 / 经验标准化。七、总结本文搭建的招聘数据爬取系统具备以下核心优势多平台适配一套框架兼容三大招聘平台适配不同的爬取模式接口监听 / 页面解析数据标准化统一字段格式实现岗位分类、薪资解析、地理编码的标准化去重可靠基于 Redis SETNX 原子操作杜绝重复数据存储规范MySQL 维度建模满足后续结构化分析需求高健壮性完善的异常捕获、字段容错、资源释放机制保证爬虫稳定运行。该系统可直接用于求职分析、行业调研等场景也可扩展为企业级的招聘数据监控平台。