简介：《数据清洗与采集论文》是一份面向大数据入门学习者、数据分析初学者及高校相关课程学生的专题文档，聚焦数据采集流程与清洗思路的系统梳理，帮助读者理解从数据源接入到存储分析的整体链路，适合课程作业、论文写作与知识查漏补缺等场景。压缩包内共1个docx文件，约313KB，内容以文字论述与架构说明为主，围绕网络爬虫的分类与应用、采集前置的信源管理、采集层的Redis缓存与任务调度、存储层的Kafka与Hadoop、HBase、Spark、Elasticsearch组合，以及辅助监控和数据质量校验等环节展开。已有66人学习，文档篇幅紧凑，便于快速建立大数据采集技术架构的全景认知，也能为后续清洗规则设计、工具选型和系统排错提供可参照的思路。
1. 从一份技术方案文档说起：数据采集与清洗这套架构到底解决什么问题
前阵子有人丢过来一份《数据清洗与采集论文.docx》，说想照着搭一套采集系统，问我值不值得看。我原本以为又是一篇堆概念的综述，翻到中间那几张架构图和四层拆解，才发现它把数据采集、数据清洗这条链路上最容易出事的几个位置都点到了：信源怎么管、任务怎么排队、同一任务怎么保证只被一个采集器处理、数据推上去之后怎么校验质量、出问题去哪套监控里查。这些恰恰是团队从"写个脚本爬网页"升级到"跑一套采集平台"时最先卡住的地方。适合正在搭数据管道的后端、做爬虫工程化的开发者，也适合想拿一份现成架构当蓝本、少画几张图的人。它不教你怎么写第一行爬虫，讲的是爬虫跑起来之后那一整套工程约束。
2. 全局架构拆解：采集前置、采集层、存储层、监控四段怎么串起来
这套架构最值得学的地方不在单个组件选得多新，而在于职责切得干净。原文把系统分成采集前置、数据采集层、数据存储层、辅助监控系统四段，每段只干一件事，段与段之间靠队列和接口解耦。很多人搭采集系统的通病是把"配置信源""调度任务""下载解析""入库"揉在一个服务里，前期跑得挺顺，信源一多、站点一反爬，整个服务就跟着卡死，改一行配置要重启全部进程。分层之后，配置改动只碰管理端，采集器崩了不影响调度，推送接口出问题也不会把已采到的数据弄丢。
2.1 四层分工与一条数据的完整旅程
先把四层的职责摆清楚，后面看代码才不会乱。
| 分层 | 模块 | 核心职责 | 承载组件 |
|---|---|---|---|
| 采集前置 | 信源管理系统 | 网站、栏目、关键词、公众号、博主的增删改查与启停 | 管理端 + 配置库 |
| 采集层 | 任务队列 | 缓存待采任务与过程状态 | Redis |
| 采集层 | 调度中心 | 按采集频率派发，保证任务唯一性 | 调度服务 + 分布式锁 |
| 采集层 | 采集器集群 | 网页下载、结构化解析、状态上报 | 多实例 Worker |
| 存储层 | 数据传输 | 质量校验、打标签、统一推送 | SpringBoot 微服务接口 |
| 存储层 | 大数据平台 | 消息缓冲、消费计算、建索引、存明细 | Kafka / Spark / ES / HBase |
| 辅助监控 | 四套子系统 | 信源、采集、服务器、数据质量 | 指标采集 + 告警 |
一条数据从生到死走的是这样一条路：运维在信源系统里配一个栏目，填上列表页规则和详情页选择器，调度中心按这个信源的采集频率生成任务，压进 Redis 队列；某个采集器抢到任务，下载页面、按配置解析出标题、发布时间、正文，调 SpringBoot 的推送接口；接口做三道校验，过了就发到 Kafka 的对应 topic；Spark 消费 topic，把检索字段写进 ES、把完整正文写进 HBase；同时链路各环节把状态回写到监控，采集失败、正文为空、时间异常都会在监控上冒出来。
这条路看着顺，但每一跳都有坑。队列里放什么、锁怎么加、校验卡在哪一层、ES 和 HBase 谁存什么，这些细节决定了系统是能跑三个月还是只能跑三天。原文给了方向，具体的落地写法得自己补，下面几章就按这条链路一段一段拆。
2.2 信源管理系统：采集任务的源头怎么建模
采集前置在原文里叫信源管理系统，管的是网站、栏目、关键词、公众号、博主，还有电子期刊、APP 客户端这类其他源。它的价值不是"能加一个网址"，而是把"采什么、怎么采、多久采一次、采回来怎么解析"变成可以增删改查的配置，让换站点不用改代码。很多人一开始把这些规则写在采集器的配置文件里，站点一多就变成几十个 yaml，谁改谁懵。
信源表我一般会拆成两张：一张 source 存信源本身，一张 source_rule 存解析规则和采集策略。拆开的好处是同一个站点下多个栏目可以共用一份下载策略，只在规则上做差异。
CREATE TABLE source ( id BIGINT PRIMARY KEY AUTO_INCREMENT, source_type VARCHAR(32) NOT NULL COMMENT 'site/channel/keyword/wechat/blog/rss/app', name VARCHAR(128) NOT NULL, entry_url VARCHAR(512) NOT NULL COMMENT '列表页或检索入口', parent_id BIGINT DEFAULT 0 COMMENT '栏目挂在站点下的父级 id', status TINYINT DEFAULT 1 COMMENT '1 启用 0 停用', frequency INT DEFAULT 3600 COMMENT '采集间隔，秒', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_type_status (source_type, status) ) COMMENT '信源主表'; CREATE TABLE source_rule ( id BIGINT PRIMARY KEY AUTO_INCREMENT, source_id BIGINT NOT NULL, page_type VARCHAR(16) NOT NULL COMMENT 'list 列表页 / detail 详情页', selector_conf JSON NOT NULL COMMENT '字段选择器配置', regex_status VARCHAR(256) DEFAULT NULL COMMENT '正则校验，命中才算采到有效内容', updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_source (source_id, page_type) ) COMMENT '解析规则表';selector_conf用 JSON 存是有意为之：字段会随站点变化，今天多一个作者字段、明天换一个正文容器，加字段不用改表结构。regex_status是原文提到的正则状态，用来兜底判断——列表页抓回来解析出 0 条链接，基本就是模板失效了，这个状态位会直接让监控报警，比等到正文全是空才反应过来早得多。frequency放在信源主表上，调度中心按它算下次派发时间，不同类型信源给的值差别很大：新闻站点几分钟一轮，公众号和博主动态半小时一轮，电子期刊一天一轮就够。
2.3 技术选型的取舍：为什么是 Redis + Kafka + Spark + ES + HBase
原文点了 Hadoop、HBase、Kafka、Spark、ES 这一串，看着像全家桶，其实每个都有明确位置，换掉任何一个都会带来实打实的麻烦。常见的取舍是这样：
| 组件 | 在这套链路里干什么 | 换掉会出什么问题 |
|---|---|---|
| Redis | 任务队列、过程状态、分布式锁 | 用数据库轮询代替，派发延迟高，压测一上来数据库先顶不住 |
| Kafka | 采集结果到计算层的缓冲 | 直连 Spark 或直写库，采集高峰和消费速度不匹配时会丢数据 |
| Spark | 消费消息、建索引、写明细 | 用单机消费者，几十万条每小时扛不住，还不好做失败重放 |
| Elasticsearch | 标题、时间、来源等检索字段 | 只留 HBase，按标题模糊检索只能全表扫 |
| HBase | 完整正文等大字段明细 | 全塞 ES，正文一长索引膨胀得厉害，查询反而变慢 |
我自己踩过一次选型的坑：早期图省事，采集结果直接写 MySQL，想着量不大先跑着。结果三千多个信源一起开，单表两千万行之后按时间范围查一次能卡十几秒，后来才补上 Kafka 和 ES，等于白做了两个月。所以看到原文把 Kafka 放在采集器和 Spark 中间，我是认同的——采集侧是突发流量，一个站点放开分页能瞬间吐出几百条，消费侧要建索引、算标签、写两份存储，速度天然慢，中间不垫一层缓冲，早晚出问题。
至于 ES 和 HBase 的分工，判断标准就一条：要按字段搜索的进 ES，要原样取回的进 HBase。标题、发布时间、来源、URL、标签这些频繁参与过滤和排序的字段建索引；正文可能几千字，甚至带图片链接，放 HBase 按行键取，ES 里只留一个关联 ID。这样 ES 的单条文档体积能压到几百字节，索引写入速度和查询延迟都会好很多。
3. 采集层落地：Redis 任务队列、调度唯一性与采集器解析
采集层是整套系统里并发压力最集中的一段，也是最容易写出重复采集和任务丢失的地方。原文强调了两件事：任务按设置的采集频率被采集，同一任务同一时间只能被一个采集器处理。这两句话翻译成实现，就是队列结构设计和分布式锁两件事。剩下的下载和解析反倒不难，难在解析规则要能改、改完要能验证。
3.1 任务队列的数据结构：List 做队列，Hash 存载荷
队列设计上最容易犯的错是把整个任务对象序列化后塞进 List，任务字段一多，列表元素就变重，几万个任务堆着，内存和出队效率都受影响。我一般的做法是队列里只放 task_id，任务详情单独用 Hash 存，出队后再按 id 取详情。另外一个细节是按信源类型分队列，新闻、公众号、博主的采集速度差得远，挤在一个队列里，慢的那种会把整个队列拖住。
import json import time import redis r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True) # 按信源类型分队列，避免长尾慢站点占满 worker QUEUE_KEY = "crawl:queue:{source_type}" TASK_KEY = "crawl:task:{task_id}" def push_task(source_type: str, task: dict) -> None: task_id = task["task_id"] # 队列只存 id，详情走 Hash，出队后再取，List 元素保持轻量 r.hset(TASK_KEY.format(task_id=task_id), mapping={ "url": task["url"], "source_id": task["source_id"], "selector_conf": json.dumps(task["selector_conf"], ensure_ascii=False), "frequency": task["frequency"], # 采集间隔，单位秒 "retry": 0, # 重试次数，超过阈值转死信 "created_at": int(time.time()), }) # 从左压入，采集器从右弹出，FIFO 顺序，同一站点不会出现新任务插队 r.lpush(QUEUE_KEY.format(source_type=source_type), task_id) # 任务详情给 7 天过期，防止异常任务长期占内存 r.expire(TASK_KEY.format(task_id=task_id), 7 * 24 * 3600)这段代码有三个参数值得留意。frequency从信源配置带过来，调度中心靠它算下一次派发时间，采集器本地不用管周期，只管干完当前这条。retry在每次失败时自增，超过阈值就推到crawl:dead死信队列，人工排查，不给它无限重试把站点打到封 IP。SELECTOR_CONF存成 JSON 字符串，采集器解析时反序列化，改解析规则只改数据库，不动代码也不重启进程。
还有一个容易忽略的点：Hash 的过期时间。任务出队执行完，crawl:task:{task_id}这条记录其实就该清了，但为了排查问题（比如某条数据为什么没入库），我会留一段时间再自动过期。真实项目里如果任务量极大，这个过期时间要调短，否则 Redis 内存会持续上涨，maxmemory一到就会开始淘汰 key，队列里的 id 一旦被淘汰而详情还在，就会出现"取到 id 查不到任务"的诡异现象。
3.2 调度唯一性：锁、processing 队列与超时补偿
"同一任务同一时间只能被一个采集器处理"这句话，实现上有两种常见思路：一种是派发时就加锁，另一种是取任务时用原子操作加上锁。我更倾向后者，因为派发侧做幂等很难覆盖所有异常，而取出侧用BRPOPLPUSH加SET NX PX的组合，能同时解决两件事——任务不丢、任务不重复。
PROCESSING_KEY = "crawl:processing" LOCK_KEY = "crawl:lock:{task_id}" def fetch_task(source_type: str, worker_id: str, timeout: int = 5): # BRPOPLPUSH 是原子的：从待采队列右弹出，同时压入 processing 队列 # 采集器中途崩溃，任务还在 processing 里，可由补偿任务重新入队 task_id = r.brpoplpush( QUEUE_KEY.format(source_type=source_type), PROCESSING_KEY, timeout=timeout, ) if not task_id: return None # 再抢一把带过期时间的锁，过期时间要明显大于单次采集耗时 ok = r.set(LOCK_KEY.format(task_id=task_id), worker_id, nx=True, px=60_000) if not ok: # 抢锁失败说明别的 worker 正在处理，把自己刚压入的这条撤掉 r.lrem(PROCESSING_KEY, 1, task_id) return None return task_id def ack_task(task_id: str, worker_id: str) -> None: # 采集成功后释放锁并从 processing 移除，两步顺序不能反 r.delete(LOCK_KEY.format(task_id=task_id)) r.lrem(PROCESSING_KEY, 1, task_id) r.delete(TASK_KEY.format(task_id=task_id))px=60_000是锁的过期时间，单位毫秒。这个值不能拍脑袋写死，得看单页采集的 P99 耗时，一般取它的两到三倍。写得小了，页面慢一点锁就提前失效，另一个采集器拿到同一个任务，重复采集就出现了；写得大了，采集器崩溃后锁要等很久才释放，任务在那段时间里一直卡着。稳妥的做法是加一个续期机制，采集器起来后开一个后台线程，每隔锁过期时间的三分之一就PEXPIRE一次，采完再主动删，这套"看门狗"逻辑在分布式任务框架里很常见。
补偿那部分容易被省掉，但生产上不能少。crawl:processing这个 List 里躺着的都是"已被取走但还没确认完成"的任务，起一个每分钟跑一次的补偿任务，扫描超过锁过期时间还在 processing 里的 id，重新压回待采队列，同时把retry加一。没有这一步，采集器一 OOM 重启，这批任务就永久丢了，监控上只看到"某站点今天数据量少了一半"，很难定位。
3.3 采集器：下载与结构化解析的可配置写法
下载和解析是采集器的主干活。原文说采集器负责网页下载、数据结构化解析、任务监控，这三件事在代码里应该分开写，别揉在一个函数里。下载只负责拿到 HTML 和状态码，解析只负责按配置抽字段，监控只在关键节点埋点。
from parsel import Selector import requests HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; Collector/1.0)"} def download(url: str, timeout: int = 10) -> str: # 超时、重试、编码统一在下载层处理，解析层拿到的永远是干净字符串 resp = requests.get(url, headers=HEADERS, timeout=timeout) resp.raise_for_status() resp.encoding = resp.apparent_encoding or "utf-8" return resp.text def parse(html: str, conf: dict) -> dict: sel = Selector(text=html) # conf 来自 source_rule.selector_conf，改规则只改配置不发版 title = sel.css(conf["title"] + "::text").get(default="").strip() time_text = sel.css(conf["time"] + "::text").get(default="") # 正文取所有匹配节点的文本，再按行拼接，兼容粗体、链接被拆散的情况 blocks = sel.css(conf["content"] + " ::text").getall() content = "\n".join(b.strip() for b in blocks if b.strip()) return {"title": title, "time_text": time_text, "content": content}parse里没有做时间格式化和字段清洗，这是故意的。解析层只负责"从页面里把东西拿出来"，拿出来的time_text可能是2024-05-12 10:30，也可能是"3小时前"，格式化和换算放到下一章的推送前校验里统一做。这样划分的好处是解析层足够薄，站点模板一变只需要换selector_conf里的 CSS 选择器，不用碰代码逻辑。
conf["content"] + " ::text"中间那个空格是有讲究的。不加空格时，选择器只会取直接文本子节点，正文里只要嵌了<a>或<strong>，那些文字就丢了，采回来的正文看起来断断续续。加上空格取所有后代文本，再用换行拼起来，结构基本能保住。当然这种做法也会把一些不该要的东西带进来，比如正文容器里夹着的"责任编辑""相关阅读"，所以selector_conf里通常还会配一个排除选择器列表，解析完再过滤一遍。
4. 存储层落地：SpringBoot 推送接口、Kafka 分发与 ES/HBase 双写
采集器把数据解析出来，到真正能查、能用，中间还隔着推送、校验、缓冲、消费、双写五步。原文在这一段给的信息最多：数据通过统一的 SpringBoot 微服务接口推送到 Kafka，同时做数据质量校验，校验发布时间、标题、正文的解析准确度，Spark 消费后把检索字段建 ES 索引、把完整信息写 HBase。这段话信息密度很高，拆开来每一步都有具体的实现要点。
4.1 统一推送接口与三道质量校验
采集器为什么不直接写 Kafka，非要绕一个 HTTP 接口？因为在接口这一层能做的事情比在采集器里做更划算：校验逻辑集中一处，所有语言的采集器都能复用；采集器和消息队列解耦，Kafka 集群做迁移、改 topic 名，采集器不用跟着改；采集器不需要持有消息队列的凭证，权限收得回来。代价是多一跳网络开销，但采集本身是秒级操作，这点开销可以接受。
@RestController @RequestMapping("/api/collect") public class CollectPushController { private static final long EARLIEST = 946656000000L; // 2000-01-01，早于这个时间基本是解析错了 @PostMapping("/push") public Result push(@RequestBody @Valid ArticleDTO dto) { // 第一道：必填与长度。标题为空、正文过短，基本是模板失效，直接打回并计数 if (!StringUtils.hasText(dto.getTitle()) || dto.getContent().length() < 50) { metrics.counter("push.reject.empty").increment(); return Result.fail("EMPTY_CONTENT"); } // 第二道：时间合理性。晚于当前时间或早于 2000 年，都是解析错了字段 long ts = dto.getPublishTime(); if (ts > System.currentTimeMillis() || ts < EARLIEST) { metrics.counter("push.reject.bad_time").increment(); return Result.fail("BAD_TIME"); } // 第三道：幂等去重。url + 标题指纹，7 天内重复的直接丢弃 String fp = DigestUtils.md5Hex(dto.getUrl() + "|" + dto.getTitle()); if (!redis.setIfAbsent("dup:" + fp, "1", Duration.ofDays(7))) { metrics.counter("push.reject.duplicate").increment(); return Result.ok("DUPLICATE"); } // 过了三道校验才进 Kafka，key 用指纹，保证同一条数据落在同一分区 kafkaTemplate.send("collect-article", fp, JSON.toJSONString(dto)); return Result.ok(fp); } }三道校验的顺序是有讲究的。必填校验最便宜，先做；时间校验次之；去重需要访问 Redis，最贵，放最后。把贵的放前面，遇到大规模模板失效时会白白打一堆 Redis 请求。
ArticleDTO里的publishTime是长整型时间戳，不是字符串。原始页面上时间格式五花八门，2024-05-12、2024/05/12 10:30、05-12、3小时前都可能出现，统一在采集器里转成时间戳再推，接口这层只判断范围。相对时间的换算我一般放在采集器侧的公共工具里，因为只有采集器知道抓取时刻，接口侧已经拿不到这个上下文了。
去重这里有个血泪经验：只按 URL 去重是不够的。同一个站点经常出现同一篇文章挂在不同栏目下、带不同查询参数的情况，URL 不一样但内容一样。加上标题做联合指纹，能挡掉大部分；如果对重复率要求更严，就得在 Spark 侧用 SimHash 做内容级相似度判断，这个放到后面二次开发的章节讲。
4.2 Kafka 消息设计与 Spark 消费侧
Kafka 的 topic 设计上，我倾向于按数据类型分，不按信源类型分。新闻、博客、公众号文章的字段结构基本一致，消费逻辑也一样，混在一个 topic 里完全没问题；但如果再细分出视频、图片、评论这些结构差异大的数据，就该另开 topic，省得消费端写一堆if type ==。分区数按吞吐量估，一般 6 到 12 个起步，采集高峰能扛住，消费端也能并行扩。
消息 key 用去重指纹，这个选择直接影响重复数据的处理。同一个 key 一定落在同一个分区，消费端按分区处理时，同一条数据不会被两个 task 同时处理，天然避免了并发写冲突。
from pyspark.sql import SparkSession, functions as F from pyspark.sql.types import StructType, StringType, LongType schema = StructType() \ .add("title", StringType()) \ .add("content", StringType()) \ .add("publishTime", LongType()) \ .add("sourceId", StringType()) \ .add("url", StringType()) spark = SparkSession.builder.appName("collect-etl").getOrCreate() raw = (spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "broker:9092") .option("subscribe", "collect-article") .option("startingOffsets", "latest") .option("maxOffsetsPerTrigger", 2000) # 单批拉取上限，防止一批太大把 ES 打满 .load()) parsed = (raw .select(F.from_json(F.col("value").cast("string"), schema).alias("d"), F.col("key").cast("string").alias("fp")) .select("fp", "d.*")) # ES 只留检索字段，正文这种大字段不往索引里塞 es_view = parsed.select("fp", "title", "publishTime", "sourceId", "url")maxOffsetsPerTrigger是这套消费里最该调的一个参数。默认不设时，Spark 一次会把当前所有积压都拉下来，赶上采集高峰，一批几十万条，下游 ES 直接吃满拒绝写入。设成两千左右，配合消费端并行度，能形成稳定的背压，积压只是延迟上升，不至于把整条链路打挂。
流式作业里还有一件事必须做：checkpointLocation。不给 checkpoint 目录，作业重启后从latest开始，重启期间的消息全丢。给了之后，重启从上次提交的 offset 继续，加上 Kafka 本身的消息保留期，配合startingOffsets策略，重放和补数都有余地。
4.3 ES 索引 mapping 与 HBase 行键设计
ES 的 mapping 最好一开始就定死，别指望动态映射。动态映射会把时间字段识别成text，把数字识别成字符串，后面想按时间范围查、按来源聚合，全得重建索引。标题字段该用中文分词器，URL 只存不索引，指纹字段用keyword。
PUT /collect_article { "settings": { "number_of_shards": 5, "number_of_replicas": 1, "refresh_interval": "30s" }, "mappings": { "properties": { "fp": { "type": "keyword" }, "title": { "type": "text", "analyzer": "ik_max_word" }, "publishTime": { "type": "date" }, "sourceId": { "type": "keyword" }, "url": { "type": "keyword", "index": false } } } }refresh_interval从默认的 1 秒调到 30 秒，是这套场景下很实用的一招。采集数据的实时性要求通常没那么高，30 秒后才可搜完全可以接受，但写入吞吐能提升一大截，批量导入时尤其明显。分片数按数据量估，单分片控制在几十 GB 以内，超过就得考虑按月建索引、配别名切换。
HBase 这边，原文只说"完整信息存入 HBASE"，实际落地最关键的是行键。行键设计不好，写入热点能把你折磨到怀疑人生。
| 行键组成 | 示例 | 作用 |
|---|---|---|
| 反转 sourceId | 3210_xxx | 让同一站点的数据不连续落在一个 Region |
| 采集时间戳 | 20240512103000 | 支持按时间范围扫描 |
| 指纹 | a1b2c3d4 | 保证同一条数据幂等写入 |
列族一般两个就够：info存标题、时间、来源这些元数据，content存正文。列族不要多，每个列族在底层对应一组 StoreFile，列族多了小文件数量上来，压缩和查询都会变慢。写入用put，同一条数据重复写会覆盖，天然幂等，配合前面的去重指纹，重复数据基本不会进来。
5. 避坑与排查：这套采集链路最容易翻车的五个位置
架构图画得再漂亮，跑起来该炸还是炸。这套链路里有几个位置属于"平时不出事，一出事就找不到原因"的类型，我把自己踩过的和原文提到但没展开的放在一起，按现象、原因、解决三个方面说清楚。
5.1 队列与调度：任务堆积、重复采集
现象一：某个类型的队列长度一直涨，采集器 CPU 却不高。
原因是队列按信源类型分得太粗。新闻类和公众号类塞在一个队列里，公众号页面加载慢、反爬严，单个任务耗时是新闻的十倍，几十个 worker 全被公众号任务占住，新闻任务在队列里排队等。监控上看不出谁慢，只看到队列堆积。
解决办法是按站点维度再分一层队列，或者给慢站点单独划一个 worker 池，用独立进程消费。同时给队列长度加告警阈值，堆积超过正常水位就报警，别等业务方来问"今天的新闻怎么少了"。
现象二：同一篇文章在 ES 里出现两三条，指纹不同。
八成是锁过期时间设小了。单次采集耗时超过锁的px值，锁自动释放，另一个 worker 抢到同一个任务，两边各推一次。如果 URL 因为分页参数不同而不一致，去重指纹就挡不住。
解决上两手抓：锁的过期时间提到单次采集 P99 的两到三倍，加看门狗续期；去重指纹从"URL + 标题"改成"URL 归一化 + 标题"，URL 里去掉utm_*、from=这类跟内容无关的查询参数。
5.2 解析与质量校验：空正文、时间错乱
现象三：采回来的正文全是导航栏和页脚文字。
选择器写得太宽，div.content在页面上匹配到多个节点，第一个恰好是模板里的公共容器。或者站点改版，正文容器从div换成了article，选择器还是老的，匹配不到就走get(default="")，最后拼出来的是一堆零散文本。
解决办法是在选择器上加层级约束，比如#main .article-body，缩小匹配范围；解析完做一次正文密度校验，正文字符数低于阈值、或者标点符号占比过低，直接判为解析失败，不进推送接口。更稳的做法是给每个信源存一份人工标注的样本页，改完选择器跑一遍回归，命中率低于 90% 就不让上线。
现象四：所有文章的发布时间都是今天。
原页面只有"3小时前""昨天 14:20"这种相对时间，解析器把采集时刻直接塞进了publishTime。时间一错，ES 里按时间排序全是刚采的，历史数据的时序完全乱掉，做趋势分析直接失真。
解决办法是在采集器里做相对时间换算，拿抓取时刻减去偏移量得到绝对时间；页面同时存在绝对时间和相对时间时，优先取绝对时间。推送接口那道"晚于当前时间或早于 2000 年"的校验能挡掉明显错误，但挡不住"全部变成今天"这种逻辑正确、语义错误的情况，所以还得在监控上加一个维度：按来源统计时间分布，某来源全部集中在今天，就该报警。
5.3 存储链路：Kafka 积压与 ES 写入拒绝
现象五：Kafka 消费延迟从几秒涨到几十分钟，ES 日志里刷429 Too Many Requests。
两个原因叠在一起。一是 Spark 单批拉得太多，maxOffsetsPerTrigger没设或者设得过大，一批几十万条同时往 ES 灌；二是 ES 的refresh_interval还是默认 1 秒，写入线程被频繁刷盘拖住，bulk 队列满了就开始拒绝。
解决办法分三层：消费侧把单批上限压到两千左右，形成稳定背压；ES 侧把refresh_interval调到 30 秒，同时把 bulk 批量大小控制在 5MB 到 10MB 之间，别一次塞几十 MB；索引侧检查副本数，写入高峰期临时把副本调成 0，写完再调回来，这个操作在批量导入场景下很常见，但记得事后恢复，不然节点挂了就没有副本可切。
这三类问题的共同点是：监控上看到的症状和真正的原因隔了好几层。所以我一般会在每个环节埋一个业务指标，队列长度、单任务耗时、推送拒绝率、消费延迟、ES 拒绝次数，五个指标摆在一块看，基本能一眼定位到是哪一段出的事。
6. 二次开发：给这套架构加信源、加去重、加验证的三步走
拿到这份方案之后要往下做，最先遇到的改造需求通常不是加功能，而是"再多接一类信源""重复数据太多要压一压""改完怎么知道有没有改坏"。这三个方向正好对应架构里的三个位置，改起来有迹可循。
6.1 新增一类信源要动哪几个地方
接一类新信源，从架构上看要动四处。信源表的source_type加一个枚举值；采集器侧加一个 adapter，负责把这类信源的入口地址转成待采任务列表，比如 RSS 直接解析 feed，APP 接口要处理签名和分页；selector_conf配一套默认模板，让新信源能直接套用；监控侧加一个维度，按新的source_type统计采集量和失败率。四处里最容易被漏掉的是监控维度，漏了之后新信源出问题，从大盘上完全看不出来，只能靠业务方反馈。
改造顺序上，我会先在测试环境跑通一条完整链路，从信源配置到 ES 里能查到数据，确认字段都齐了再放量。放量时不要一次全开，先把frequency调大，跑一天看看失败率和数据量是否符合预期，再逐步收紧。
6.2 增量去重与指纹计算
现在的去重是 URL 加标题的 MD5，能挡住大部分重复，但挡不住"同一篇文章换个标题重发""正文被转载站改了开头结尾"这些情况。要压得更狠，就得引入内容指纹。
| 方案 | 计算方式 | 适用场景 | 代价 |
|---|---|---|---|
| MD5(URL + 标题) | 精确匹配 | 同一 URL 重复采集 | 极低，但换 URL 就失效 |
| SimHash(正文) | 分词后加权降维成 64 位 | 转载、轻微改写 | 需要分词，计算量中等 |
| 正文前 N 字 MD5 | 取正文前 500 字哈希 | 快速粗筛 | 低，但改写开头就失效 |
我一般用两级：先查 MD5 精确去重，命不中再算 SimHash，与最近 7 天的指纹做汉明距离比较，距离小于 3 判为重复。SimHash 的计算放在 Spark 消费侧做，别放在推送接口里，接口那层扛不住分词的开销。指纹存 Redis 的 Set 或者单独一张表，按时间分区，过期的定期清掉。
6.3 改造后的验证清单
改完之后怎么确认没改坏，我总结了几条固定要走的：
# 1. 单信源全链路验证：配一条测试信源，看能否从队列一路走到 ES curl -s "http://es:9200/collect_article/_search?q=sourceId:test_source" | jq '.hits.total' # 2. 失败重放验证：手动往队列压一条错误 URL，观察 retry 自增和死信队列 redis-cli lrange crawl:dead 0 10 # 3. 去重验证：同一条数据连推三次，ES 里应该只有一条 for i in 1 2 3; do curl -X POST http://api:8080/api/collect/push -d @sample.json; done # 4. 消费延迟验证：看 Kafka 消费组的 lag 是否在分钟级以内 kafka-consumer-groups.sh --describe --group collect-etl这四条跑完，基本能覆盖"数据进得来、错的数据挡得住、重复的拦得下、消费不积压"四件事。少跑一条，上线后就可能在某处翻车。
这套架构我照着改过两轮，最大的教训是：每次加信源、改选择器、调去重策略，都别只看"新数据对不对"，一定要回头看一眼老信源的数据量有没有掉。有一次我改了解析层的公共工具函数，新接的站点一切正常，结果一个跑了半年的老站点悄无声息地采空了三天，因为它的页面结构刚好命中了新逻辑里的一个边界条件，而监控只按总量告警，总量被新站点的增量盖住了。从那以后我每次动公共代码，都强制把核心信源的日采集量拉出来对一遍，掉了立刻回滚。希望帮到你。
本文还有配套的精品资源，点击获取
