简介《数据清洗与采集论文》是一份面向大数据、数据工程方向学习者及课程论文写作者的专题文档重点解决数据采集流程梳理、技术架构理解与写作素材组织等问题。文档共1个docx文件压缩包约313KB篇幅集中、便于检索。内容从数据采集的意义与网络爬虫的通用、聚焦分类讲起依次拆解采集前置、数据采集层、数据存储层和辅助监控系统并延伸到数据源类型、批处理与流处理工具、分布式存储及分析系统等环节对Redis任务队列、Kafka消息中间件、Hadoop/HBase/Spark/ES等组件的作用也有说明配有技术架构总结与数据质量校验思路。已有66人学习适合用于课程作业参考、知识框架搭建、面试复习也可为毕业设计或项目方案中的采集模块选型与架构表述提供借鉴。1. 采集系统上线三个月脏数据比没数据更难受某公司做行业资讯聚合第一版爬虫是三天赶出来的脚本跑了两周运营打开后台想拉一份周报发现库里四成是重复标题发布时间清一色是脚本运行当天正文里还混着导航栏和版权声明。数据不是没有是不能用。后来翻到这份《数据清洗与采集论文.docx》它没给任何能直接跑的源码给的是一张完整的工程地图数据采集的信源怎么管、任务怎么排队、解析完的数据在哪一层做清洗和质量校验、监控该盯哪几个指标。这份文档适合两类人——正在从零搭采集系统的后端以及手上已经堆了一摞爬虫脚本、但数据一进库就开始发臭的团队。它把链路拆成采集前置、采集层、存储层、辅助监控四段每段都写清了职责边界和失败时的表现这点比多数教程实在。2. 四层架构拆解从信源配置到 ES 索引的一条完整数据流2.1 四层的职责边界和一次请求的走向文档里画的那个结构抽象成一句话就是谁配、谁采、谁存、谁盯。采集前置是信源管理系统负责网站、栏目、关键词、公众号、博主、电子期刊、APP 客户端这些采集源的增删改查运营在后台改一个采集频率链路末端立刻生效数据采集层是干活的部分Redis 放任务队列和过程数据调度中心决定谁在什么时候采多个采集器实例负责下载和结构化解析数据存储层做两件事一是把解析结果按统一格式推给 Kafka二是在 Spark 里做质量校验、打标签最后标题时间正文进 ES 建索引、全量信息进 HBase辅助监控系统不属于主链路但贯穿全程分信源监控、采集监控、服务器监控、数据质量校验四个子系统。这个分层最值得学的不是技术选型是切分方式。它把「配置」和「执行」彻底隔开了——信源配置存在关系库里采集器只认队列里的任务不知道任务从哪来采集器也不直接写数据库所有数据统一走一个推送接口进 Kafka。这么做的好处是每个环节都能单独重启、单独扩容。我见过太多脚本式采集配置硬编码在代码里改一个站点的选择器要重新发版而按这个结构改选择器只是改一行模板记录。常见的错法是把调度逻辑写在采集器里每个采集器各自算下一次什么时候跑最后同一站点被采了八遍。从数据源类型看文档分了结构化关系库、NoSQL和非结构化日志、社交媒体、音视频对应的采集方式完全不同。结构化源一般走定时增量拉取靠 update_time 或者自增 ID 做游标非结构化源才需要爬虫这套东西。存储选型那段也说得直白数据量过 TB 之后传统关系库顶不住分布式存储成了必选项Hadoop 管离线、HBase 管明细、ES 管检索、Kafka 管缓冲各司其职。2.2 信源管理系统的字段设计整个链路的配置源头文档只说了「方便运维人员对采集源增删改查」没给表结构。真落地的时候这张表是整条链路的配置中心字段少一个后面就得改代码。我一般会这么建CREATE TABLE collect_source ( id BIGINT PRIMARY KEY AUTO_INCREMENT, source_type VARCHAR(16) NOT NULL COMMENT WEB/COLUMN/KEYWORD/WECHAT/BLOGGER/JOURNAL/APP, name VARCHAR(128) NOT NULL COMMENT 信源名称运营可见, entry_url VARCHAR(512) NOT NULL COMMENT 入口地址关键词类存搜索词, charset VARCHAR(16) DEFAULT AUTO COMMENT AUTO 走响应头meta 探测, template_id BIGINT NOT NULL COMMENT 解析模板版本决定选择器集合, cron_expr VARCHAR(64) NOT NULL COMMENT 采集频率调度中心解析, priority TINYINT DEFAULT 5 COMMENT 1最高 9最低决定入哪个队列, status TINYINT DEFAULT 1 COMMENT 1启用 0暂停 2仅监控不采集, fail_count INT DEFAULT 0 COMMENT 连续失败次数达到阈值自动降频, last_run_at DATETIME NULL COMMENT 上次成功采集时间, UNIQUE KEY uk_type_url (source_type, entry_url(255)) );逐字段说下为什么这么设计。template_id单独拎出来指向一张模板表是为了让「选择器变更」和「信源变更」解耦——同一个新闻站的列表页模板可能被 30 个栏目信源共用改一次全生效而且模板表可以留版本历史出问题能回滚。charset设成 AUTO 是有血泪经验的国内不少老站点响应头声明 GBK 但实际内容是 UTF-8硬编码必然乱码只能靠 meta 探测加二次校验兜底。priority不是排序用的是分队列用的。高优先级信源比如竞品官网进collect:queue:p1低优先级历史归档站点进collect:queue:p5采集器实例按队列权重分配消费能力避免一个慢站点把队列堵死。fail_count配合status做自动降频连续失败 10 次把 cron 从 10 分钟拉到 2 小时连续失败 50 次直接置为暂停并告警防止调度中心对着一个已经下线的站点空转。uk_type_url这个唯一键是防手滑的运营重复录入同一个栏目插入时直接报错比事后去重便宜得多。提示entry_url建前缀索引(255)是因为 MySQL 的索引长度限制如果你的库开了更大的innodb_large_prefix可以直接全字段索引但采集 URL 超过 255 字符的情况极少前缀够用。2.3 一次采集任务从入队到入库的完整时序信源配好之后任务是怎么流动的文档里没写时序但按它描述的四层结构可以推出来实际实现基本是这五步。第一步调度中心按固定周期常见做法是每分钟一次扫一遍collect_source里status1且last_run_at cron_expr已到期的记录为每条生成一个任务任务 ID 用source_id 时间窗拼出来比如src_1024_202406120900。这个 ID 设计很关键它天然带时间窗去重——同一信源同一个十分钟窗口内只会有一个任务即使调度中心因为抖动跑了两遍也不会重复入队。第二步任务 ID 按优先级LPUSH进对应的 Redis 队列同时把任务详情写进collect:task:state:{task_id}这个 Hash 里字段包括状态、重试次数、开始时间、worker 标识。第三步采集器实例用BRPOP阻塞式从队列取任务取到之后先抢分布式锁抢到才执行抢不到就放回队列等下一轮。第四步采集器下载页面、按模板解析出标题、发布时间、正文、作者、来源把结构化结果组织成统一格式调 SpringBoot 的推送接口。第五步推送接口做基础校验后写 Kafka同时把任务状态从RUNNING改成SUCCESS或FAILED失败的任务按退避策略重新入队。Spark 那边消费 Kafka做完深度校验和打标签分别写 ES 和 HBase。这五步里最容易出问题的是第四步和第五步之间——采集器解析成功但推送失败如果没做本地暂存这条数据就丢了。文档提到「数据传输负责将采集器解析出来的内容通过统一微服务接口推送」隐含的要求是采集器必须能感知推送结果并重试不能一推了之。3. 采集层实操Redis 任务队列与调度中心的并发唯一性控制3.1 Redis 的键设计和队列优先级拆分文档说 Redis 用来「缓存采集任务队列、过程数据」一句话带过实际上一套键名规划不好后面排查能把人逼疯。我按职责把键分四类命名统一用collect:前缀加业务域。键名类型用途过期策略collect:queue:p{n}List按优先级拆分的任务队列不设过期collect:task:state:{taskId}Hash任务状态、重试次数、workerId7 天collect:task:lock:{taskId}String执行权锁NX 抢占300 秒 心跳续期collect:source:meta:{sourceId}Hash信源配置缓存避免每次查库10 分钟collect:dedup:{yyyyMMdd}Set当日已采 URL 指纹秒级去重2 天按优先级拆队列的理由前面提过这里补一句实际效果p1 队列的消费超时设 30 秒p5 设 120 秒同样的采集器实例数高优先级信源的实际延迟能压到 1 分钟以内。如果只有一个队列一个卡死的视频站能拖慢整条链路运营看到的就是「昨天还能采的站点今天全停了」而服务器 CPU 才 5%这种玄学问题查起来最费时间。URL 指纹去重那个 Set 是为了挡掉列表页翻页导致的重复入队。做法很简单入队前对规范化后的 URL 取 MD5SADD返回 0 就说明今天采过直接跳过。注意是「入队前」而不是「入库前」越靠前挡掉越省资源。规范化要处理三件事去掉 utm 之类跟踪参数、统一协议和域名大小写、去掉锚点。缓存信源配置这一层也别省。调度中心每分钟扫一次假设有 5000 个信源每次都查库就是 5000 次查询加上采集器取任务时可能还要读配置数据库压力不小。用 Hash 缓存 10 分钟配置变更时主动删 key读的时候 miss 再回源成本能降一个数量级。3.2 唯一性保证锁、心跳和任务状态机「同一任务同一时间只能被一个采集器处理」这句话文档只给了结论落地要靠分布式锁。最朴素的写法是这样// 采集器消费循环BRPOP 阻塞出队 NX 锁抢占执行权 public CollectTask poll(String queueKey) { // 1. 阻塞 5 秒取任务避免空轮询把 Redis QPS 打满 String taskId redis.brpop(5, queueKey); if (taskId null) return null; // 2. NX 保证只有一个 worker 拿到执行权EX 是兜底 TTL String lockKey collect:task:lock: taskId; boolean locked redis.set(lockKey, workerId, NX, EX, 300); if (!locked) { return null; // 已被别人持有本轮放弃 } // 3. 抢到锁后写状态标记为 RUNNING redis.hset(collect:task:state: taskId, status, RUNNING); redis.hset(collect:task:state: taskId, worker, workerId); return loadTask(taskId); }这段的关键在参数。BRPOP的超时设 5 秒而不是 0永久阻塞是为了让 worker 有机会响应优雅停机信号否则进程被 kill 的时候还挂在阻塞调用上任务状态永远停在出队那一刻。EX 300是兜底过期正常路径上必须起一个续期线程每 60 秒把 TTL 重置回 300 秒直到任务结束主动删除锁。不续期的后果很直接一个大页面加图片下载跑了 6 分钟锁在第 5 分钟自动过期另一个 worker 抢到锁开始重复采同一个任务被处理两遍重试次数还被算乱了。任务状态机建议只设五个状态PENDING入队、RUNNING抢到锁、SUCCESS、FAILED可重试、DEAD超过最大重试次数。状态流转只能单向SUCCESS和DEAD是终态。文档里监控子系统说「监控每个采集任务的状态以便对异常任务、数据漏采等情况进行排查」排查的前提就是状态机足够简单运维看一眼状态分布就知道是队列堵了还是解析挂了。锁的粒度也要想清楚。按任务加锁是最细的同一个信源的不同时间窗任务可以并行如果按信源加锁能防止对同一站点并发请求过快被封但会降低吞吐。折中做法是两层锁信源级用 Redis 的令牌桶限速比如每秒最多 2 个请求任务级用上面的 NX 锁。文档没提限速但没有限速的采集系统活不过一个月这条属于必须自己补的。3.3 调度频率、退避重试和采集器的解析职责调度中心的频率控制有个容易忽略的点cron_expr是「期望频率」不是「保证频率」。如果队列积压任务实际执行时间会晚于计划时间这时候调度中心不该继续往队列里塞新任务否则积压越来越深。常见做法是给每个信源维护一个「在途任务数」超过 1 就不再生成新任务等前一个出结果再说。文档说「保证任务按设置的采集频率被采集」严格说是做不到的能保证的是「不早于这个频率」。重试策略我一般用阶梯退避第 1 次失败 1 分钟后重试第 2 次 5 分钟第 3 次 30 分钟第 4 次 2 小时第 5 次直接进死信队列并把fail_count加一。区分错误类型很重要——连接超时、5xx 这类可以重试404、403 这类重试没意义直接进死信并触发信源监控告警。文档的监控子系统里有「信源系统监控监控网站、栏目、公众号、博主等状态保证其正常访问」这个告警就是它的输入端。采集器本身的职责文档写得很清楚网页下载、数据结构化解析、任务监控。拆开看是三块。下载层要做连接池复用、超时控制连接 5 秒、读取 15 秒是个合理起点、UA 和 Cookie 管理、编码探测解析层靠模板列表页用 CSS 选择器或者 XPath 定位条目详情页要处理正文抽取——简单的用选择器复杂的站点选择器经常失效得fallback 到正文密度算法任务监控层负责上报耗时、状态、失败原因这些数据直接喂给辅助监控系统。解析结果一定要带parseTemplateVersion。模板改了之后历史数据的解析口径和新的不一致排查「为什么上周的标题带尾巴这周不带」的时候没有版本号就只能靠猜。这个字段成本几乎为零价值极大。4. 清洗与落库SpringBoot 推送接口到 Kafka、Spark、ES 的完整链路4.1 统一数据模型和推送接口的约定数据存储层的第一道关口是推送接口。文档说「通过统一的 SpringBoot 微服务接口推送到 Kafka 中间件同时对数据的质量进行校验主要校验发布时间、标题、正文等解析的准确度」。统一的意思是所有采集器的输出必须是同一个结构不管来源是网站、公众号还是 APP。消息体我一般定成这样{ taskId: src_1024_202406120900, sourceId: 1024, sourceType: COLUMN, url: https://example.com/news/123.html, title: 某行业季度报告发布, publishTime: 2024-06-12 08:47:00, author: , content: 正文纯文本已去除标签和多余空白, contentHash: 9f2c4a1e8b7d..., parseTemplateVersion: 7, collectTime: 2024-06-12 09:00:12, retryCount: 0 }字段解释一下。content存纯文本不存 HTML是为了让下游 Spark 少做一次解析而且 HTML 里的样式类名经常变存下来反而干扰比对。contentHash是对「标题 正文前 2000 字」做 MD5这是去重的主键比 URL 去重可靠得多——同一篇文章经常有多个入口 URL靠 URL 根本挡不住。publishTime允许为空字符串不能填采集时间这个坑后面单独说。taskId和sourceId保留是为了溯源出问题时能顺着这两个字段反查到具体哪次调度、哪个信源配置。parseTemplateVersion前面说过是解析口径的版本标记。retryCount让下游能识别出这条数据是重试来的Spark 里可以对重试数据放宽某些校验比如时间戳容差避免正常数据被误判。接口层面有两个约定值得单独强调。一是幂等同一个taskId重复推送必须能被识别接口侧用 Redis 做SETNX挡掉或者干脆不管让下游按contentHash去重。二是背压采集器推送要设超时和重试上限接口返回 429 就退避不能无限重试——Kafka 写满的时候接口会变慢采集器如果死等整个采集层就雪崩了。这两条文档没写但凡是生产环境跑过的都知道躲不开。4.2 Kafka topic 规划和 Spark 消费的清洗规则topic 按数据量拆别都塞一个。常见拆法是按信源类型分collect.web网站和栏目、collect.social公众号、博主、collect.app客户端、期刊。分区数按峰值吞吐估单分区 510 MB/s 是个保守值3 个 topic 每个 6 分区起步够用一阵。消息 key 用contentHash保证同一篇文章落到同一分区下游做窗口去重方便。Spark 消费侧的核心工作是清洗。文档提到要校验「发布时间、标题、正文的解析准确度」还提到打标签和特定信源监控。我把校验规则写成这样import re, hashlib from datetime import datetime, timedelta def normalize(doc, collect_time): errors [] title (doc.get(title) or ).strip() content (doc.get(content) or ).strip() # 标题长度 5~200不能是站点名或纯导航词 if not (5 len(title) 200): errors.append(title_length) # 正文少于 100 字基本是解析失败大概率抽到了导航栏 if len(content) 100: errors.append(content_too_short) # 发布时间多格式尝试失败就置空绝不回填采集时间 raw (doc.get(publishTime) or ).strip() pub None for fmt in (%Y-%m-%d %H:%M:%S, %Y/%m/%d %H:%M, %Y年%m月%d日 %H:%M): try: pub datetime.strptime(raw, fmt); break except ValueError: continue if raw and pub is None: errors.append(publish_time_unparsable) # 未来时间或早于 2000 年都视为解析错误 if pub and (pub collect_time timedelta(hours1) or pub.year 2000): errors.append(publish_time_out_of_range) pub None doc[publishTime] pub.strftime(%Y-%m-%d %H:%M:%S) if pub else doc[contentHash] hashlib.md5((title content[:2000]).encode()).hexdigest() doc[qualityFlags] ,.join(errors) return doc逻辑上分三层。第一层是必填校验标题和正文长度是硬指标正文不足 100 字直接判为解析失败——这个阈值不是拍脑袋正常新闻正文很少低于这个数而导航栏加版权声明的长度恰好在这个区间。第二层是发布时间解析多格式尝试之后仍然失败就把字段置空并打标记绝对不能用采集时间代替。这一条是全文最有价值的一句话因为发布时间一旦被污染后面所有的时间维度分析全是错的而且错得很隐蔽重新采集才能修复。第三层是异常值过滤未来时间、2000 年以前的时间基本都是解析串位置空比留着强。qualityFlags这个字段是给运营看的。带 flag 的数据不进 ES 主索引进一个「待复核」索引运营抽样看几条就能判断是模板失效还是站点改版比翻日志快得多。文档里「数据质量校验根据异常数据反查信源等配置」说的就是这个闭环。4.3 ES 索引设计和 HBase 的冷热分工Spark 处理完往两边写。ES 存标题、时间、正文、来源、标签这些检索和展示需要的字段HBase 存全量原始数据包括 HTML供事后回溯。这个分工的逻辑很实际ES 存全量 HTML 会让索引膨胀几倍查询还慢HBase 便宜、能存大字段但没法做全文检索。ES 的 mapping 一定要手写别用动态映射。动态映射会把publishTime识别成 text之后按天聚合直接报错得重建索引才能救。我用的 mapping 骨架{ mappings: { properties: { title: { type: text, analyzer: ik_max_word, fields: { raw: { type: keyword } } }, content: { type: text, analyzer: ik_max_word }, publishTime: { type: date, format: yyyy-MM-dd HH:mm:ss||epoch_millis }, sourceId: { type: keyword }, contentHash: { type: keyword }, qualityFlags:{ type: keyword }, collectTime: { type: date, format: yyyy-MM-dd HH:mm:ss } } } }title同时给 text 和 keyword 两种类型是为了既能全文搜又能精确聚合去重contentHash设 keyword 是因为要做terms聚合查重sourceId用 keyword 不用 long是为了兼容部分信源 ID 带前缀的情况。索引按天或按月滚动用别名指向最新索引老索引关掉副本降成本。HBase 的 rowkey 用contentHash反转加时间前缀反转是为了避免写热点——同一时间段的数据 hash 前缀分布随机region 能均匀扛住写入。列族拆两个info存结构化字段raw存原始 HTML 和响应头后者压缩率开高一点。监控这块文档列的四个子系统基本覆盖了必须盯的维度落到具体指标上大概是信源监控看各站点最近一次成功采集时间和连续失败次数采集监控看队列长度、任务成功率、平均耗时、死信数量服务器监控看 CPU、内存、磁盘、进程存活质量校验看qualityFlags的分布变化。四个维度里最容易漏的是最后一个前三个挂了会立刻报警质量是慢慢劣化的等运营发现数据不对可能已经污染了半个月的数据。5. 避坑排查上线后最容易翻车的五个环节和对应解法5.1 队列和锁任务卡住、积压、重复采现象一后台看到一批任务状态永远停在 RUNNING队列长度不降。原因基本是采集器进程被 kill 或者 OOM 之后锁没释放任务状态也没回写而队列里的任务已经被BRPOP弹出去了。解决分两步锁必须带 TTL 加心跳续期worker 启动时扫一遍自己名下的 RUNNING 任务做补偿超过 2 倍超时时间还没结束的直接改回 PENDING 重新入队。另外给 worker 注册优雅停机钩子收到信号先把当前任务处理完或者主动释放锁别硬扛。现象二队列堆积了几万条服务器 CPU 却只有 5%。原因是单个 worker 用同步阻塞的方式处理任务一个响应慢的站点比如要等 30 秒才返回的政府站把整个消费线程占住。解法是拉取和处理分离主线程只负责BRPOP和抢锁把任务丢进本地线程池执行线程池大小按「平均耗时 × 目标 QPS」估算。同时给每个信源配独立的信号量防止慢站点占满所有线程。文档没写这块但并发模型选错后面加机器都救不回来。现象三同一篇文章在库里出现三遍contentHash还不一样。原因通常不是去重失效是正文里混进了随机内容——推荐位、阅读数、时间戳。同一篇文章两次抓取正文有细微差异hash 自然不同。解决是先做正文清洗再算 hash去掉连续的空白和换行、剔除常见的干扰段落含「责任编辑」「本文来源」「点击查看更多」的行再取前 2000 字。如果还压不住降到 SimHash 加汉明距离阈值 3用 Spark 侧做近似去重。这条属于典型的「以为去重写好了其实没写对」。5.2 解析和落库时间污染、大消息、映射翻车现象四报表里所有文章的发布时间都是同一天。原因很明确解析器抽不到发布时间时用了new Date()兜底或者取的是列表页的渲染时间。这条错误的可怕之处在于它不会报错数据看起来完整只有做时间序列分析时才发现全是平的。解决是把空值当合法值处理抽不到就留空并打publish_time_missing标记让质量看板去暴露问题而不是用假数据填坑。另外时间解析要限定合理区间未来时间、2000 年以前一律判为异常。现象五Kafka 偶发拒绝消息日志里是 RecordTooLargeException。公众号长文加上内嵌的图片 base64一条消息能到几 MB超过 broker 默认的 1 MB 限制。治标的做法是调大max.request.size和message.max.bytes但治标不治本broker 的内存压力会变大。我一般改成正文超过 512 KB 就把内容落对象存储消息里只带一个 keySpark 消费时按需拉取。这样消息体恒定在几 KB吞吐稳定重放也方便。还有一个和 ES 相关的索引用了动态映射sourceId被识别成 long后来某个信源 ID 带了前缀变成字符串写不进去还报类型冲突。解法是上线前就把 mapping 固化所有可能做聚合和精确匹配的字段统一用 keyword新增字段走索引模板而不是让 ES 自己猜。这类问题在索引建立当天发现最省事等积累了半年数据再改只能重建。6. 二次开发把文档里的架构图改造成能跑的最小闭环拿到这份文档别一上来就照着搭 Hadoop 全家桶那是给已经有大数据平台的团队准备的。我建议先砍到最小闭环一个 Redis、一个采集器、一个本地文件或 SQLite 落地把「信源配置 → 入队 → 抢锁 → 下载解析 → 清洗 → 去重」这条链路跑通再逐个替换组件。替换顺序按痛点来通常是先加 Kafka 解耦采集和入库再加 ES 支撑检索HBase 最后上因为它是成本最高、收益最晚的一环。# 解析回归脚本每次改模板前先跑一遍看通过率有没有掉 import json, sys from extractor import parse_detail # 你的解析实现 cases json.load(open(golden_samples.json, encodingutf-8)) failed [] for c in cases: doc parse_detail(open(c[html], encodingutf-8).read(), c[url]) for field, expect in c[expect].items(): if doc.get(field) ! expect: failed.append((c[name], field, expect, doc.get(field))) print(f通过 {len(cases) - len(failed)}/{len(cases)}) for f in failed[:10]: print(f) sys.exit(1 if failed else 0)这个脚本的价值在于给解析模板上保险。golden_samples.json里每条记录一个信源、一份存档 HTML、一份期望输出字段比对要求完全相等。模板改动后先跑一遍通过率掉了就地回滚比上线之后靠运营反馈快得多。样本量不用大每个信源两条一条正常、一条异常页面就够关键是长期维护——信源改版的时候顺手把新页面的样本补进去半年后你会感谢自己。验证方法上还有个小技巧给每个信源加一个「影子模式」新模板先不落库只把解析结果和旧模板的结果做 diff 打到日志里观察一两天没问题再切正式。这个机制在信源批量改版的时候能救命我吃过一次亏三十个栏目一次性换模板有个站点的标题选择器写错了一位跑到第二天才发现标题全是列表页的栏目名那批数据只能重采。从那以后我每次改解析模板都强制走一遍黄金样本集加影子比对不管改动多小。希望帮到你。本文还有配套的精品资源点击获取