简介一套基于Python爬虫和MapReduce分析的招聘信息大数据可视化系统毕业设计源码面向计算机相关专业学生、教师及初期开发者适用毕业设计、课程设计、作业演示或进阶学习。系统从招聘网站抓取数据经MapReduce分析处理利用Flask构建可视化大屏覆盖数据采集、清洗、分析、展示全流程可完整展示招聘市场行情与需求分布。压缩包共207个文件、约17.13MB主要类型包括Python核心逻辑、HTML/CSS/JS前端交互、JSON配置、SQLite数据库及部署文档gif动图用于演示运行效果rar中补充数据资料方便离线使用。目前已有334人学习浏览。项目已在macOS/Windows/Linux多平台运行验证并获导师认可、评审分95分附带部署文档和全部数据资料目录结构清晰便于复现和二次开发。在此基础上可进一步扩展行业薪酬、技能热词、岗位需求等分析模块也能作为企业招聘数据可视化的参考原型。1. 招聘信息大数据可视化系统从爬虫到 MapReduce 的技术闭环一个招聘网站的页面里藏着城市薪资分布、岗位技能要求、学历门槛这些高价值信息但如果靠人工逐个翻页整理两个站点就能把人耗垮。这个标题所描述的毕业设计本质上是一条完整的数据流水线Python 爬虫负责持续采集原始招聘信息MapReduce 负责把半结构化文本变成可聚合的统计指标最后通过可视化图表把统计结果呈现出来。整个系统覆盖了“采集 → 清洗 → 存储 → 分布式计算 → 可视化展示”五个环节适合正在做大数据方向毕设、想补全工程实践细节的同学阅读也适合数据分析岗位的从业者参考其离线分析链路设计。它不是一个单体爬虫脚本而是一个能回答“某城市 Java 岗位平均薪资是多少”这类问题的完整基础设施。2. Python 爬虫采集层Requests 抓取、内容解析与数据落库2.1 为什么选 Requests 而非 Scrapy做招聘数据采集很多参考教程会直接推荐 Scrapy 框架。但在这个系统里我一般会建议先用 Requests 把流程跑通原因有三个。第一Requests 的代码路径短一个函数完成请求、超时控制、编码处理排查问题时能直接看到网络交互的完整上下文第二招聘网站的页面结构相对规整内容以列表页加详情页为主不需要 Scrapy 那样的中间件链来做复杂调度第三毕设答辩时面试官问“这个请求为什么这样写”Requests 的代码每一行都能解释清楚而 Scrapy 的很多行为被框架封装掉了。如果后续采集规模扩大再把 Requests 的采集函数改造成 Scrapy 的 Spider成本并不高。数据落库方面招聘信息属于典型的宽表结构用 MySQL 存储比 MongoDB 更直观也方便后续用 Sqoop 或直接导出 CSV 喂给 HDFS。表结构设计需要预留分析字段这一步不要节省。CREATE TABLE job_post ( id INT PRIMARY KEY AUTO_INCREMENT, job_title VARCHAR(128) NOT NULL COMMENT 职位名称, company_name VARCHAR(128) COMMENT 公司名称, city VARCHAR(32) COMMENT 工作城市, salary_min INT COMMENT 薪资下限单位K, salary_max INT COMMENT 薪资上限单位K, experience VARCHAR(32) COMMENT 经验要求如 3-5年, education VARCHAR(32) COMMENT 学历要求, job_desc TEXT COMMENT 职位描述用于文本挖掘, fetch_date DATE COMMENT 采集日期, INDEX idx_city_job (city, job_title) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;把薪资拆成salary_min和salary_max而不是直接存字符串是为了后续 MapReduce 做数值聚合时不需要反复做正则解析。job_desc保留原始文本用于技能词频统计。2.2 Requests 爬虫最小可运行代码下面这段代码可以看作整个采集层的最小闭环请求列表页、解析职位卡片、翻页循环。目标站点统一称为“目标招聘站点”你需要根据实际站点调整 CSS 选择器。import requests import time import pymysql from bs4 import BeautifulSoup HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml, Accept-Language: zh-CN,zh;q0.9, } def fetch_page(city_code, page_no, keywordpython): url fhttps://example-job-site.com/search?city{city_code}kw{keyword}page{page_no} resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text def parse_jobs(html): soup BeautifulSoup(html, lxml) jobs [] for item in soup.select(.job-card): title_tag item.select_one(.job-title) salary_text item.select_one(.salary).get_text(stripTrue) salary_min, salary_max parse_salary(salary_text) jobs.append({ job_title: title_tag.get_text(stripTrue), company_name: item.select_one(.company).get_text(stripTrue), city: item.select_one(.city).get_text(stripTrue), salary_min: salary_min, salary_max: salary_max, experience: item.select_one(.exp).get_text(stripTrue), education: item.select_one(.edu).get_text(stripTrue), }) return jobs def parse_salary(text): # 输入示例15K-25K、10K-15K·13薪 clean text.split(·)[0].replace(K, ).replace(k, ) parts clean.split(-) if len(parts) 2: return int(parts[0]), int(parts[1]) return None, None逻辑说明fetch_page负责一次 HTTP 请求返回解码后的 HTML 文本parse_jobs从 HTML 中提取职位卡片每个卡片对应一条结构化记录。parse_salary专门处理薪资文本把带单位的字符串转成数值。参数说明里值得注意的有三点。第一timeout10是必带的否则某个响应慢的站点会让采集线程长时间挂起第二resp.encoding resp.apparent_encoding用于规避中文站点因响应头缺失导致的乱码第三HEADERS里的User-Agent是请求头的核心大量站点对默认 Python UA 直接拒绝响应。2.3 增量采集与去重策略首次全量采集之后后续任务只关心新增职位和已关闭职位。常见做法是维护一张job_fetch_log表记录每次采集的任务批次用fetch_date做按日分区。去重则以job_title company_name city为业务唯一键入库前先查一次 MySQL。def save_jobs(cursor, jobs, fetch_date): sql (INSERT INTO job_post (job_title, company_name, city, salary_min, salary_max, experience, education, job_desc, fetch_date) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)) rows [(j[job_title], j[company_name], j[city], j[salary_min], j[salary_max], j[experience], j[education], j.get(job_desc, ), fetch_date) for j in jobs] cursor.executemany(sql, rows)城市场景下反爬对抗的重心不在请求头伪装而在采集节奏。两次请求之间加 1 到 3 秒随机延时比堆砌请求头更有效。这个延时既是为了降低目标站点压力也是为了让系统在长时间运行时不触发访问频率限制。3. MapReduce 统计分析从 MySQL 导出到 HDFS 与 Streaming 作业3.1 数据从 MySQL 到 HDFS 的迁移链路爬虫落库之后的数据还在 MySQLMapReduce 作业的输入必须是 HDFS 上的文件。最直接的路径是先用 SQL 查出分析所需的宽表导出为 CSV再传到 HDFS。这一步不需要引入 Sqoop减少环境依赖。mysql -uadmin -p --batch --skip-column-names \ -e SELECT city, job_title, salary_min, salary_max, education, job_desc FROM job_post WHERE fetch_date2025-01-01 \ /data/job_20250101.csv hdfs dfs -mkdir -p /warehouse/job_analysis hdfs dfs -put /data/job_20250101.csv /warehouse/job_analysis/ hdfs dfs -ls /warehouse/job_analysis/命令说明--batch让 MySQL 以非交互模式输出--skip-column-names去掉表头行避免 MapReduce 把字段名当成数据记录。CSV 文件按日期命名这样 HDFS 上的目录天然支持按时间分区扫描。3.2 用 Hadoop Streaming 跑 Python Mapper 与 Reducer标题里明确写了 MapReduce但实际做数据统计时不需要写 Java 代码。Hadoop Streaming 允许把任意可执行文件当作 Mapper 和 ReducerPython 脚本通过标准输入读取数据、通过标准输出写出键值对。这个机制非常适合招聘信息这种半结构化文本的处理。先看 Mapper 的职责解析 CSV 行按城市维度输出“城市 薪资区间样本”。#!/usr/bin/env python3 # mapper.py import sys def parse_csv_line(line): # 简单按逗号切分实际数据含逗号时需要 csv 模块处理 parts line.strip().split(,) if len(parts) 5: return None, None city parts[0] try: salary_min int(parts[2]) salary_max int(parts[3]) except ValueError: return None, None return city, (salary_min, salary_max) for raw_line in sys.stdin: city, salary parse_csv_line(raw_line) if city and salary[0] and salary[1]: avg_salary (salary[0] salary[1]) / 2 print(f{city}\t{avg_salary})Mapper 的输入是 HDFS 上 CSV 文件中的一行输出以制表符分隔。\t左边是 key、右边是 valueHadoop 会按 key 排序后把同一个 key 的所有 value 交给同一个 Reducer 处理。Reducer 端负责计算每个城市的平均薪资与岗位样本数#!/usr/bin/env python3 # reducer.py import sys current_city None total_salary 0.0 sample_count 0 for raw_line in sys.stdin: line raw_line.strip() if not line: continue city, salary line.split(\t, 1) try: salary_val float(salary) except ValueError: continue if current_city is None: current_city city if city ! current_city: avg total_salary / sample_count if sample_count else 0 print(f{current_city}\t{sample_count}\t{avg}) current_city city total_salary 0.0 sample_count 0 total_salary salary_val sample_count 1 if current_city is not None: avg total_salary / sample_count if sample_count else 0 print(f{current_city}\t{sample_count}\t{avg})代码逻辑说明current_city用于追踪 key 的变化一旦 key 切换就把上一个城市的累计值输出并重置。这里没有用字典缓存全部数据因为 Streaming 模式下每个进程只处理一部分 key内存开销可控。3.3 作业提交命令与核心参数hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \ -files /home/user/mapper.py,/home/user/reducer.py \ -mapper python3 mapper.py \ -reducer python3 reducer.py \ -input /warehouse/job_analysis/job_20250101.csv \ -output /warehouse/job_analysis/result_20250101 \ -numReduceTasks 4-files把本地脚本分发到集群每个节点脚本依赖的程序不会自动打包所以直接用python3解释器-numReduceTasks设置 Reducer 数量这个值不是越大越好参考值是输入数据块数的 0.95 倍数据量小的时候设置 4 到 8 个即可。输出目录必须是 HDFS 上不存在的路径否则作业直接报错。跑完作业后看结果用一条命令验证产出hdfs dfs -cat /warehouse/job_analysis/result_20250101/part-r-00000如果要做学历要求分布Mapper 的输出 key 换成学历_城市的组合键Reducer 聚合逻辑不变。MapReduce 的设计模式在这里体现得很清楚把复杂的统计拆成“局部计算 全局合并”两步。提示实际在伪分布式环境跑时job_desc字段如果包含换行符CSV 的行会被切断。导出时用REPLACE(job_desc, \n, )清洗后再导出。4. 可视化层ECharts 仪表盘与 Flask 聚合接口4.1 可视化选型为什么是 ECharts 而不是 BI 工具招聘数据可视化的核心诉求是“交互式筛选”比如按城市切换查看薪资中位数按技能关键词过滤岗位分布。市面上的 BI 工具确实能拖拽生成图表但对毕业设计来说前后端联调是绕不开的加分项。ECharts 的地图、词云、漏斗图都是现成的而且支持数据动态更新。数据流组织方式MapReduce 的结果落在 HDFS 上可视化后端不能直接读 HDFS常见做法是把 part-r-00000 这类结果文件导出到 MySQL 的分析表中Flask 提供查询接口前端图表异步调用。4.2 Flask 后端聚合接口from flask import Flask, jsonify, request import pymysql app Flask(__name__) DB_CONFIG { host: localhost, user: analyst, password: yourpassword, database: job_analysis, charset: utf8mb4, } def query_all(sql, argsNone): conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: cursor.execute(sql, args) return cursor.fetchall() finally: conn.close() app.route(/api/salary_by_city) def salary_by_city(): rows query_all( SELECT city, sample_count, avg_salary FROM city_salary_summary ORDER BY sample_count DESC ) return jsonify([{ city: r[0], count: r[1], avg_salary: round(r[2], 1), } for r in rows]) app.route(/api/job_wordcloud) def job_wordcloud(): keyword request.args.get(keyword, ) sql (SELECT job_title, COUNT(*) AS cnt FROM job_post WHERE job_title LIKE %s GROUP BY job_title ORDER BY cnt DESC LIMIT 50) rows query_all(sql, (f%{keyword}%,)) return jsonify([{name: r[0], value: r[1]} for r in rows]) if __name__ __main__: app.run(host0.0.0.0, port8080, debugFalse)query_all封装了连接建立、查询、关闭的完整流程避免每个接口重复写数据库样板代码。注意 Flask 的debugTrue只能在开发环境开部署时打开会让控制台暴露堆栈并降低并发性能。4.3 ECharts 地图与词云的前端配置地图是招聘系统最常用的图表先看柱状图的数据绑定方式async function loadCitySalary() { const resp await fetch(/api/salary_by_city); const data await resp.json(); const cities data.map(item item.city); const values data.map(item item.avg_salary); const chart echarts.init(document.getElementById(salaryChart)); chart.setOption({ tooltip: { trigger: axis }, xAxis: { type: category, data: cities }, yAxis: { type: value, name: 平均薪资(K) }, series: [{ name: 城市平均薪资, type: bar, data: values, itemStyle: { color: #5470c6 }, }] }); } loadCitySalary();如果把xAxis.type改成value、series.type改成scatter就能快速切换成散点图用于观察“经验要求”和“薪资”的相关性。可视化层的价值在于把 MapReduce 跑出来的数字转化成可判断的分布形态而不是追求图表的装饰复杂度。提示ECharts 的地图组件需要引入中国地图的 geo JSON 数据这部分数据要提前下载放到本地static/目录否则直接引用 CDN 地址在离线环境会加载失败。5. 伪分布式环境下的部署验证与增量爬取技巧5.1 全链路数据一致性验证系统上线后第一件事是验证“MySQL 里的记录数”和“HDFS 上的统计结果”是否对得上。常见做法是分别统计三个环节的行数任何一处不一致都说明清洗或序列化有问题。# 第一步MySQL 侧统计 mysql -uadmin -p -e SELECT COUNT(*), COUNT(DISTINCT city) FROM job_post WHERE fetch_date2025-01-01; # 第二步HDFS 侧统计 CSV 行数 hdfs dfs -cat /warehouse/job_analysis/job_20250101.csv | wc -l # 第三步验证 MapReduce 输出 key 是否完整 hdfs dfs -cat /warehouse/job_analysis/result_20250101/part-r-* | awk -F \t {sum $2; print $1, $2} END {print TOTAL, sum}第三种方式输出的城市列表和 MySQL 中COUNT(DISTINCT city)比较城市数量一致且 SUM 等于总行数就可以判定全链路无数据缺失。这个验证脚本建议封装成 shell 文件每次修改爬虫解析规则后重跑一遍。5.2 增量爬取的幂等设计招聘网站每天都有新职位上线和旧职位下架全量爬取浪费资源且会给目标站点造成访问压力。我习惯用“业务唯一键 fetch_date”的双重机制做增量。采集批次开始时先查询库中已有的业务键集合Requests 抓到的职位如果业务键已存在就跳过插入只记录“职位仍然在架”的标记。这样重复采集同一页不会产生脏数据。MapReduce 侧对应地把输入文件按日期隔离分析哪一天的数据就用哪一天的 CSV。5.3 旧数据回刷与统计口径修正爬虫解析规则一旦调整历史数据往往需要重算。最稳妥的方案不是直接改代码后重跑整个流水线而是维护一个rule_version字段每次解析规则变更这个字段加一。MapReduce 作业支持-D mapreduce.input.fileinputformat.input.dir.recursivetrue配合 HDFS 目录后缀按版本过滤这样只有受影响的历史分区会被重新计算。最后一个值得注意的技巧是 CSV 文件中job_desc的文本清洗把所有换行符提前替换为空格。MapReduce 按行读取的机制决定了任何嵌入换行符都会把一条记录劈成两行导致统计口径错乱。清洗动作要放在 MySQL 导出阶段而不是 Hadoop 脚本里因为后者只能看到已经被切开的数据流无法还原被换行符破坏的记录边界。本文还有配套的精品资源点击获取 招聘数据采集与MapReduce分析可视化系统实战
简介一套基于Python爬虫和MapReduce分析的招聘信息大数据可视化系统毕业设计源码面向计算机相关专业学生、教师及初期开发者适用毕业设计、课程设计、作业演示或进阶学习。系统从招聘网站抓取数据经MapReduce分析处理利用Flask构建可视化大屏覆盖数据采集、清洗、分析、展示全流程可完整展示招聘市场行情与需求分布。压缩包共207个文件、约17.13MB主要类型包括Python核心逻辑、HTML/CSS/JS前端交互、JSON配置、SQLite数据库及部署文档gif动图用于演示运行效果rar中补充数据资料方便离线使用。目前已有334人学习浏览。项目已在macOS/Windows/Linux多平台运行验证并获导师认可、评审分95分附带部署文档和全部数据资料目录结构清晰便于复现和二次开发。在此基础上可进一步扩展行业薪酬、技能热词、岗位需求等分析模块也能作为企业招聘数据可视化的参考原型。1. 招聘信息大数据可视化系统从爬虫到 MapReduce 的技术闭环一个招聘网站的页面里藏着城市薪资分布、岗位技能要求、学历门槛这些高价值信息但如果靠人工逐个翻页整理两个站点就能把人耗垮。这个标题所描述的毕业设计本质上是一条完整的数据流水线Python 爬虫负责持续采集原始招聘信息MapReduce 负责把半结构化文本变成可聚合的统计指标最后通过可视化图表把统计结果呈现出来。整个系统覆盖了“采集 → 清洗 → 存储 → 分布式计算 → 可视化展示”五个环节适合正在做大数据方向毕设、想补全工程实践细节的同学阅读也适合数据分析岗位的从业者参考其离线分析链路设计。它不是一个单体爬虫脚本而是一个能回答“某城市 Java 岗位平均薪资是多少”这类问题的完整基础设施。2. Python 爬虫采集层Requests 抓取、内容解析与数据落库2.1 为什么选 Requests 而非 Scrapy做招聘数据采集很多参考教程会直接推荐 Scrapy 框架。但在这个系统里我一般会建议先用 Requests 把流程跑通原因有三个。第一Requests 的代码路径短一个函数完成请求、超时控制、编码处理排查问题时能直接看到网络交互的完整上下文第二招聘网站的页面结构相对规整内容以列表页加详情页为主不需要 Scrapy 那样的中间件链来做复杂调度第三毕设答辩时面试官问“这个请求为什么这样写”Requests 的代码每一行都能解释清楚而 Scrapy 的很多行为被框架封装掉了。如果后续采集规模扩大再把 Requests 的采集函数改造成 Scrapy 的 Spider成本并不高。数据落库方面招聘信息属于典型的宽表结构用 MySQL 存储比 MongoDB 更直观也方便后续用 Sqoop 或直接导出 CSV 喂给 HDFS。表结构设计需要预留分析字段这一步不要节省。CREATE TABLE job_post ( id INT PRIMARY KEY AUTO_INCREMENT, job_title VARCHAR(128) NOT NULL COMMENT 职位名称, company_name VARCHAR(128) COMMENT 公司名称, city VARCHAR(32) COMMENT 工作城市, salary_min INT COMMENT 薪资下限单位K, salary_max INT COMMENT 薪资上限单位K, experience VARCHAR(32) COMMENT 经验要求如 3-5年, education VARCHAR(32) COMMENT 学历要求, job_desc TEXT COMMENT 职位描述用于文本挖掘, fetch_date DATE COMMENT 采集日期, INDEX idx_city_job (city, job_title) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;把薪资拆成salary_min和salary_max而不是直接存字符串是为了后续 MapReduce 做数值聚合时不需要反复做正则解析。job_desc保留原始文本用于技能词频统计。2.2 Requests 爬虫最小可运行代码下面这段代码可以看作整个采集层的最小闭环请求列表页、解析职位卡片、翻页循环。目标站点统一称为“目标招聘站点”你需要根据实际站点调整 CSS 选择器。import requests import time import pymysql from bs4 import BeautifulSoup HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml, Accept-Language: zh-CN,zh;q0.9, } def fetch_page(city_code, page_no, keywordpython): url fhttps://example-job-site.com/search?city{city_code}kw{keyword}page{page_no} resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text def parse_jobs(html): soup BeautifulSoup(html, lxml) jobs [] for item in soup.select(.job-card): title_tag item.select_one(.job-title) salary_text item.select_one(.salary).get_text(stripTrue) salary_min, salary_max parse_salary(salary_text) jobs.append({ job_title: title_tag.get_text(stripTrue), company_name: item.select_one(.company).get_text(stripTrue), city: item.select_one(.city).get_text(stripTrue), salary_min: salary_min, salary_max: salary_max, experience: item.select_one(.exp).get_text(stripTrue), education: item.select_one(.edu).get_text(stripTrue), }) return jobs def parse_salary(text): # 输入示例15K-25K、10K-15K·13薪 clean text.split(·)[0].replace(K, ).replace(k, ) parts clean.split(-) if len(parts) 2: return int(parts[0]), int(parts[1]) return None, None逻辑说明fetch_page负责一次 HTTP 请求返回解码后的 HTML 文本parse_jobs从 HTML 中提取职位卡片每个卡片对应一条结构化记录。parse_salary专门处理薪资文本把带单位的字符串转成数值。参数说明里值得注意的有三点。第一timeout10是必带的否则某个响应慢的站点会让采集线程长时间挂起第二resp.encoding resp.apparent_encoding用于规避中文站点因响应头缺失导致的乱码第三HEADERS里的User-Agent是请求头的核心大量站点对默认 Python UA 直接拒绝响应。2.3 增量采集与去重策略首次全量采集之后后续任务只关心新增职位和已关闭职位。常见做法是维护一张job_fetch_log表记录每次采集的任务批次用fetch_date做按日分区。去重则以job_title company_name city为业务唯一键入库前先查一次 MySQL。def save_jobs(cursor, jobs, fetch_date): sql (INSERT INTO job_post (job_title, company_name, city, salary_min, salary_max, experience, education, job_desc, fetch_date) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)) rows [(j[job_title], j[company_name], j[city], j[salary_min], j[salary_max], j[experience], j[education], j.get(job_desc, ), fetch_date) for j in jobs] cursor.executemany(sql, rows)城市场景下反爬对抗的重心不在请求头伪装而在采集节奏。两次请求之间加 1 到 3 秒随机延时比堆砌请求头更有效。这个延时既是为了降低目标站点压力也是为了让系统在长时间运行时不触发访问频率限制。3. MapReduce 统计分析从 MySQL 导出到 HDFS 与 Streaming 作业3.1 数据从 MySQL 到 HDFS 的迁移链路爬虫落库之后的数据还在 MySQLMapReduce 作业的输入必须是 HDFS 上的文件。最直接的路径是先用 SQL 查出分析所需的宽表导出为 CSV再传到 HDFS。这一步不需要引入 Sqoop减少环境依赖。mysql -uadmin -p --batch --skip-column-names \ -e SELECT city, job_title, salary_min, salary_max, education, job_desc FROM job_post WHERE fetch_date2025-01-01 \ /data/job_20250101.csv hdfs dfs -mkdir -p /warehouse/job_analysis hdfs dfs -put /data/job_20250101.csv /warehouse/job_analysis/ hdfs dfs -ls /warehouse/job_analysis/命令说明--batch让 MySQL 以非交互模式输出--skip-column-names去掉表头行避免 MapReduce 把字段名当成数据记录。CSV 文件按日期命名这样 HDFS 上的目录天然支持按时间分区扫描。3.2 用 Hadoop Streaming 跑 Python Mapper 与 Reducer标题里明确写了 MapReduce但实际做数据统计时不需要写 Java 代码。Hadoop Streaming 允许把任意可执行文件当作 Mapper 和 ReducerPython 脚本通过标准输入读取数据、通过标准输出写出键值对。这个机制非常适合招聘信息这种半结构化文本的处理。先看 Mapper 的职责解析 CSV 行按城市维度输出“城市 薪资区间样本”。#!/usr/bin/env python3 # mapper.py import sys def parse_csv_line(line): # 简单按逗号切分实际数据含逗号时需要 csv 模块处理 parts line.strip().split(,) if len(parts) 5: return None, None city parts[0] try: salary_min int(parts[2]) salary_max int(parts[3]) except ValueError: return None, None return city, (salary_min, salary_max) for raw_line in sys.stdin: city, salary parse_csv_line(raw_line) if city and salary[0] and salary[1]: avg_salary (salary[0] salary[1]) / 2 print(f{city}\t{avg_salary})Mapper 的输入是 HDFS 上 CSV 文件中的一行输出以制表符分隔。\t左边是 key、右边是 valueHadoop 会按 key 排序后把同一个 key 的所有 value 交给同一个 Reducer 处理。Reducer 端负责计算每个城市的平均薪资与岗位样本数#!/usr/bin/env python3 # reducer.py import sys current_city None total_salary 0.0 sample_count 0 for raw_line in sys.stdin: line raw_line.strip() if not line: continue city, salary line.split(\t, 1) try: salary_val float(salary) except ValueError: continue if current_city is None: current_city city if city ! current_city: avg total_salary / sample_count if sample_count else 0 print(f{current_city}\t{sample_count}\t{avg}) current_city city total_salary 0.0 sample_count 0 total_salary salary_val sample_count 1 if current_city is not None: avg total_salary / sample_count if sample_count else 0 print(f{current_city}\t{sample_count}\t{avg})代码逻辑说明current_city用于追踪 key 的变化一旦 key 切换就把上一个城市的累计值输出并重置。这里没有用字典缓存全部数据因为 Streaming 模式下每个进程只处理一部分 key内存开销可控。3.3 作业提交命令与核心参数hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \ -files /home/user/mapper.py,/home/user/reducer.py \ -mapper python3 mapper.py \ -reducer python3 reducer.py \ -input /warehouse/job_analysis/job_20250101.csv \ -output /warehouse/job_analysis/result_20250101 \ -numReduceTasks 4-files把本地脚本分发到集群每个节点脚本依赖的程序不会自动打包所以直接用python3解释器-numReduceTasks设置 Reducer 数量这个值不是越大越好参考值是输入数据块数的 0.95 倍数据量小的时候设置 4 到 8 个即可。输出目录必须是 HDFS 上不存在的路径否则作业直接报错。跑完作业后看结果用一条命令验证产出hdfs dfs -cat /warehouse/job_analysis/result_20250101/part-r-00000如果要做学历要求分布Mapper 的输出 key 换成学历_城市的组合键Reducer 聚合逻辑不变。MapReduce 的设计模式在这里体现得很清楚把复杂的统计拆成“局部计算 全局合并”两步。提示实际在伪分布式环境跑时job_desc字段如果包含换行符CSV 的行会被切断。导出时用REPLACE(job_desc, \n, )清洗后再导出。4. 可视化层ECharts 仪表盘与 Flask 聚合接口4.1 可视化选型为什么是 ECharts 而不是 BI 工具招聘数据可视化的核心诉求是“交互式筛选”比如按城市切换查看薪资中位数按技能关键词过滤岗位分布。市面上的 BI 工具确实能拖拽生成图表但对毕业设计来说前后端联调是绕不开的加分项。ECharts 的地图、词云、漏斗图都是现成的而且支持数据动态更新。数据流组织方式MapReduce 的结果落在 HDFS 上可视化后端不能直接读 HDFS常见做法是把 part-r-00000 这类结果文件导出到 MySQL 的分析表中Flask 提供查询接口前端图表异步调用。4.2 Flask 后端聚合接口from flask import Flask, jsonify, request import pymysql app Flask(__name__) DB_CONFIG { host: localhost, user: analyst, password: yourpassword, database: job_analysis, charset: utf8mb4, } def query_all(sql, argsNone): conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: cursor.execute(sql, args) return cursor.fetchall() finally: conn.close() app.route(/api/salary_by_city) def salary_by_city(): rows query_all( SELECT city, sample_count, avg_salary FROM city_salary_summary ORDER BY sample_count DESC ) return jsonify([{ city: r[0], count: r[1], avg_salary: round(r[2], 1), } for r in rows]) app.route(/api/job_wordcloud) def job_wordcloud(): keyword request.args.get(keyword, ) sql (SELECT job_title, COUNT(*) AS cnt FROM job_post WHERE job_title LIKE %s GROUP BY job_title ORDER BY cnt DESC LIMIT 50) rows query_all(sql, (f%{keyword}%,)) return jsonify([{name: r[0], value: r[1]} for r in rows]) if __name__ __main__: app.run(host0.0.0.0, port8080, debugFalse)query_all封装了连接建立、查询、关闭的完整流程避免每个接口重复写数据库样板代码。注意 Flask 的debugTrue只能在开发环境开部署时打开会让控制台暴露堆栈并降低并发性能。4.3 ECharts 地图与词云的前端配置地图是招聘系统最常用的图表先看柱状图的数据绑定方式async function loadCitySalary() { const resp await fetch(/api/salary_by_city); const data await resp.json(); const cities data.map(item item.city); const values data.map(item item.avg_salary); const chart echarts.init(document.getElementById(salaryChart)); chart.setOption({ tooltip: { trigger: axis }, xAxis: { type: category, data: cities }, yAxis: { type: value, name: 平均薪资(K) }, series: [{ name: 城市平均薪资, type: bar, data: values, itemStyle: { color: #5470c6 }, }] }); } loadCitySalary();如果把xAxis.type改成value、series.type改成scatter就能快速切换成散点图用于观察“经验要求”和“薪资”的相关性。可视化层的价值在于把 MapReduce 跑出来的数字转化成可判断的分布形态而不是追求图表的装饰复杂度。提示ECharts 的地图组件需要引入中国地图的 geo JSON 数据这部分数据要提前下载放到本地static/目录否则直接引用 CDN 地址在离线环境会加载失败。5. 伪分布式环境下的部署验证与增量爬取技巧5.1 全链路数据一致性验证系统上线后第一件事是验证“MySQL 里的记录数”和“HDFS 上的统计结果”是否对得上。常见做法是分别统计三个环节的行数任何一处不一致都说明清洗或序列化有问题。# 第一步MySQL 侧统计 mysql -uadmin -p -e SELECT COUNT(*), COUNT(DISTINCT city) FROM job_post WHERE fetch_date2025-01-01; # 第二步HDFS 侧统计 CSV 行数 hdfs dfs -cat /warehouse/job_analysis/job_20250101.csv | wc -l # 第三步验证 MapReduce 输出 key 是否完整 hdfs dfs -cat /warehouse/job_analysis/result_20250101/part-r-* | awk -F \t {sum $2; print $1, $2} END {print TOTAL, sum}第三种方式输出的城市列表和 MySQL 中COUNT(DISTINCT city)比较城市数量一致且 SUM 等于总行数就可以判定全链路无数据缺失。这个验证脚本建议封装成 shell 文件每次修改爬虫解析规则后重跑一遍。5.2 增量爬取的幂等设计招聘网站每天都有新职位上线和旧职位下架全量爬取浪费资源且会给目标站点造成访问压力。我习惯用“业务唯一键 fetch_date”的双重机制做增量。采集批次开始时先查询库中已有的业务键集合Requests 抓到的职位如果业务键已存在就跳过插入只记录“职位仍然在架”的标记。这样重复采集同一页不会产生脏数据。MapReduce 侧对应地把输入文件按日期隔离分析哪一天的数据就用哪一天的 CSV。5.3 旧数据回刷与统计口径修正爬虫解析规则一旦调整历史数据往往需要重算。最稳妥的方案不是直接改代码后重跑整个流水线而是维护一个rule_version字段每次解析规则变更这个字段加一。MapReduce 作业支持-D mapreduce.input.fileinputformat.input.dir.recursivetrue配合 HDFS 目录后缀按版本过滤这样只有受影响的历史分区会被重新计算。最后一个值得注意的技巧是 CSV 文件中job_desc的文本清洗把所有换行符提前替换为空格。MapReduce 按行读取的机制决定了任何嵌入换行符都会把一条记录劈成两行导致统计口径错乱。清洗动作要放在 MySQL 导出阶段而不是 Hadoop 脚本里因为后者只能看到已经被切开的数据流无法还原被换行符破坏的记录边界。本文还有配套的精品资源点击获取