简介《数据清洗与采集论文》是一份面向大数据入门学习者、数据分析初学者及高校相关课程学生的专题文档聚焦数据采集流程与清洗思路的系统梳理帮助读者理解从数据源接入到存储分析的整体链路适合课程作业、论文写作与知识查漏补缺等场景。压缩包内共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 的对应 topicSpark 消费 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(host127.0.0.1, port6379, db0, decode_responsesTrue) # 按信源类型分队列避免长尾慢站点占满 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_idtask_id), mapping{ url: task[url], source_id: task[source_id], selector_conf: json.dumps(task[selector_conf], ensure_asciiFalse), frequency: task[frequency], # 采集间隔单位秒 retry: 0, # 重试次数超过阈值转死信 created_at: int(time.time()), }) # 从左压入采集器从右弹出FIFO 顺序同一站点不会出现新任务插队 r.lpush(QUEUE_KEY.format(source_typesource_type), task_id) # 任务详情给 7 天过期防止异常任务长期占内存 r.expire(TASK_KEY.format(task_idtask_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_typesource_type), PROCESSING_KEY, timeouttimeout, ) if not task_id: return None # 再抢一把带过期时间的锁过期时间要明显大于单次采集耗时 ok r.set(LOCK_KEY.format(task_idtask_id), worker_id, nxTrue, px60_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_idtask_id)) r.lrem(PROCESSING_KEY, 1, task_id) r.delete(TASK_KEY.format(task_idtask_id))px60_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, headersHEADERS, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding or utf-8 return resp.text def parse(html: str, conf: dict) - dict: sel Selector(texthtml) # 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); } // 过了三道校验才进 Kafkakey 用指纹保证同一条数据落在同一分区 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实际落地最关键的是行键。行键设计不好写入热点能把你折磨到怀疑人生。行键组成示例作用反转 sourceId3210_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 直接解析 feedAPP 接口要处理签名和分页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?qsourceId: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这四条跑完基本能覆盖数据进得来、错的数据挡得住、重复的拦得下、消费不积压四件事。少跑一条上线后就可能在某处翻车。这套架构我照着改过两轮最大的教训是每次加信源、改选择器、调去重策略都别只看新数据对不对一定要回头看一眼老信源的数据量有没有掉。有一次我改了解析层的公共工具函数新接的站点一切正常结果一个跑了半年的老站点悄无声息地采空了三天因为它的页面结构刚好命中了新逻辑里的一个边界条件而监控只按总量告警总量被新站点的增量盖住了。从那以后我每次动公共代码都强制把核心信源的日采集量拉出来对一遍掉了立刻回滚。希望帮到你。本文还有配套的精品资源点击获取