大数据采集系统架构设计：从爬虫调度到Kafka与Spark存储全链路实践

简介在大数据链路中数据采集与清洗是决定数据质量的上游环节直接影响后续分析的准确性。一个健壮的采集系统需要涵盖信源管理、任务调度、消息缓冲和存储落地等多个层次。本文将拆解四层采集架构重点讨论如何利用Redis实现分布式任务锁与队列去重通过Kafka削峰填谷解耦采集与存储再由Spark消费数据并分别写入Elasticsearch与HBase兼顾查询性能与全量存储。同时针对调度重复、数据漏采、消息积压等生产高频问题给出工程化排查思路。文中还包含可参考的Python、Kafka与Spark代码片段适合正在构建或优化采集系统的开发者参考。1. 数据采集这关过不去后面清洗、分析全是白用功做数据这行的人大概都有过这种体验辛辛苦苦把模型调好了结果喂进去的数据有一半是空字段、重复ID、乱码正文最后得出的结论自己都不敢信。数据采集和清洗在整个大数据链路里属于最不起眼、但翻车率最高的环节。很多团队把精力放在存储、计算、可视化上却忽略了一个事实采集端决定了数据的天花板。你后面用再好的清洗规则也救不回来采集时丢失的字段。这份文档讲的是头条、博客、公众号这类信源的完整采集架构从采集前置管理到任务调度、再到数据落库和监控是一套能直接照着搭的工程方案。适合正在做爬虫项目、或者想从零搭一套采集系统的开发者。不是讲概念的是讲怎么把数据稳稳当当拿回来的。2. 四层架构拆解采集前置、采集层、存储层、监控系统各自在扛什么活完整的大数据采集系统不是写几个爬虫脚本那么简单。文档里的架构图很清晰整个系统分四层采集前置负责管信源、采集层负责干活、存储层负责把数据送进仓库、监控系统负责盯着所有环节不出问题。这四层缺一个系统都会变成黑匣子。2.1 采集前置不只是增删改查是信源的生命周期管理采集前置本质上是一个信源管理系统。它管的不只是网站列表还有栏目、关键词、微信公众号、博主主页这些采集源。这个设计有一个很关键的好处把采集目标和采集逻辑解耦。运营人员不用碰代码直接在前置系统里加一个网站、配一个栏目任务就会被调度层拉起来。我一般会建议在这个环节加上信源状态字段不只是启用/停用还要记录最近一次采集时间、最近一次成功时间、最近一次失败原因。为什么因为信源是会变的网站改版了、公众号改名了、博主删了旧文章这些都是常态。如果前置系统里没有这些状态采集层会一遍又一遍地跑失败任务浪费资源还查不出原因。关键词采集是另一个容易被忽视的点。文档里提到关键词用于搜索引擎采集这块的难点在于关键词的增删和启停必须是实时的。运营今天想追一个热点加一个关键词系统得马上开始采而不是等下一个调度周期。所以关键词配置的变更要直接推到任务队列里而不是等调度器轮询。这里我踩过坑当时关键词变更走了配置中心生效延迟五分钟热点新闻早就被别人采走了。2.2 数据采集层Redis 缓存队列和任务调度中心是怎么配合的采集层是整个系统的核心执行单元由 Redis 缓存平台、任务调度中心和多个采集器组成。Redis 在这里不只是缓存它实际上是任务队列的存储介质。采集前置产生的任务先进入 Redis 队列调度中心负责按频率把任务分发给采集器。任务调度的关键在于唯一性控制。文档里明确写着同一任务同一时间只能被一个采集器处理这是防止重复采集的基础。实现方式一般是 Redis 的 SETNX 命令任务 ID 作为 key谁抢到锁谁执行。还有个细节是任务去重同一个 URL 如果已经在队列里就不应该再被添加。这里我一般会用一个已消费 URL 的集合做判断但这个集合不能无限增长要注意定期清理。采集器拿到任务后要做三件事网页下载、结构化解析、任务状态上报。网页下载要注意超时设置和重试机制用 requests 或 scrapy 都行但代理池要做好。结构化解析是技术含量最高的部分同一篇文章在不同网站可能结构完全不同所以解析规则最好做成可配置的而不是硬编码在代码里。# 采集器任务处理的核心逻辑 def process_task(task): task_id task[id] # 用 Redis SETNX 保证同一任务同一时间只被一个采集器处理 lock_acquired redis_client.set(ftask_lock:{task_id}, 1, nxTrue, ex300) if not lock_acquired: return # 任务已被其他采集器处理直接跳过 try: # 1. 网页下载设置超时和重试 html download_page(task[url], timeout10, retries3) # 2. 结构化解析提取标题、正文、发布时间 data parse_content(html, task[parse_rule]) # 3. 校验数据完整性 if validate_data(data): # 4. 推送到 Kafka交给存储层处理 producer.send(raw_data, keytask[url], valuedata) # 5. 上报任务状态 task_reporter.success(task_id, len(data[content])) else: task_reporter.failed(task_id, 数据校验不通过) except Exception as e: # 任务失败要记录原因方便监控系统反查 task_reporter.failed(task_id, str(e)) finally: # 释放任务锁让后续任务可以处理 redis_client.delete(ftask_lock:{task_id})这段逻辑里有几个关键参数要注意。nxTrue表示只有当 key 不存在时才能设置成功这是实现互斥锁的核心。ex300是锁的超时时间防止采集器崩溃后锁永远不释放。超时时间取 5 分钟是因为正常情况下一个采集任务不可能超过这个时长。重试次数设为 3 次比较合理超过 3 次基本可以断定是信源的问题应该标记失败而不是继续死磕。还有一个容易被忽略的点validate_data 这一步。很多采集系统只做解析不做校验导致脏数据直接进 Kafka。校验至少包括三个字段标题不能为空、正文不能为空、发布时间不能为空或离谱比如 1970 年。这一步做扎实了后面存储层的数据质量压力会小很多。3. 存储链路落地SpringBoot 微服务推 KafkaSpark 消费写 HBase 和 ES采集器解析完数据后接下来是数据存储层。文档里的链路很清晰采集器通过统一的 SpringBoot 微服务接口把数据推送到 Kafka 中间件Spark 消费 Kafka 中的数据把标题、时间、正文创建 ES 索引供查询同时把完整信息存入 HBase。这条链路是大数据采集系统的标准做法。3.1 为什么中间要隔一层 Kafka而不是采集器直接写数据库直接写数据库看起来省事但后果很严重。采集器的数量是动态的运营加了一个采集源可能就有新的采集器被拉起来瞬间产生大量写入请求。如果这些请求全部直接打到 HBase 上存储层的压力会非常不稳定而且在采集高峰期很容易把数据库打挂。Kafka 在这里起的是削峰填谷的作用。采集器只负责往 Kafka 里推数据不关心数据最终被谁消费。Spark 消费端可以根据自己的处理能力调整消费速度数据在 Kafka 里缓冲不会丢失也不会积压到不可控。这个设计还有一个好处采集器和存储层之间没有耦合任何一端升级都不影响另一端。Kafka 的 topic 设计也有讲究。我一般建议按数据源类型分 topic比如 news_topic、blog_topic、wechat_topic。为什么因为不同类型的数据消费逻辑不同新闻可能需要做关键词标签公众号文章可能需要做原创检测。分 topic 之后Spark 消费端可以针对不同类型的消息走不同的处理逻辑避免一个 topic 里混着各种格式的数据解析的时候还要先判断类型。# 创建 Kafka topic按数据源类型隔离 kafka-topics.sh --create \ --bootstrap-server kafka-01:9092 \ --topic news_topic \ --partitions 6 \ --replication-factor 2分区数设 6 是一个经验值。分区数越多消费并行度越高但也不完全是越多越好。分区数超过 Kafka 节点数时副本分布会让部分分区处于冷状态反而增加管理开销。一般经验是一个分区对应 1-2 个消费者实例分区数是节点数的 2-3 倍比较合理。副本因子设 2 是底线生产环境最好设 3Kafka 节点挂掉一台不影响数据完整性和消费连续性。3.2 Spark 消费写双存储的取舍ES 管查询、HBase 管全量别搞反Spark 消费 Kafka 数据后要做两件事创建 ES 索引供业务查询同时把完整信息存入 HBase。这个设计有一个核心考量ES 擅长全文检索和条件查询但存储成本高不适合存全量数据HBase 擅长海量数据的随机读写但查询能力弱。所以把需要查询的字段放进 ES把完整数据存 HBase各司其职。这里有一个关键决策点哪些字段进 ES哪些字段只进 HBase。我的建议是标题、发布时间、信源名称、URL 必须进 ES因为业务查询基本都围绕这几个字段正文内容要不要进 ES取决于业务需求如果要做全文检索就进如果只需要按标题查就只在 HBase 里存。正文是非常占存储空间的一个 5000 字的网页正文可能就有 10KB 以上全部进 ES 的成本会很高。# Spark 消费 Kafka 消息写 ES 和 HBase from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col spark SparkSession.builder \ .appName(data-consumer) \ .config(spark.es.nodes, es-01) \ .getOrCreate() # 从 Kafka 读取数据流 df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka-01:9092) \ .option(subscribe, news_topic) \ .option(startingOffsets, earliest) \ .load() # 解析 JSON 数据 parsed df.select( from_json(col(value).cast(string), schematitle STRING, content STRING, publish_time STRING, source STRING, url STRING).alias(data) ).select(data.*) # 清洗过滤空标题和空正文 cleaned parsed.filter(col(title) ! col(content) ! ) # 写入 ES供业务查询 cleaned.writeStream \ .format(org.elasticsearch.spark.sql) \ .option(es.resource, news_index) \ .option(checkpointLocation, /tmp/es-checkpoint) \ .start() # 写入 HBase保存全量数据 def write_hbase(batch_df, batch_id): batch_df.foreachPartition(write_to_hbase) cleaned.writeStream \ .foreachBatch(write_hbase) \ .option(checkpointLocation, /tmp/hbase-checkpoint) \ .start()这段代码里有两个 Stream 同时消费同一个 Kafka topic这在结构上是允许的但要理解这样做的含义。ES 和 HBase 各有一个独立的消费通道互不干扰一个挂了另一个还能继续工作。startingOffsets设为earliest意味着 Kafka 里积压的数据会被重新消费这在系统重启后特别重要能保证数据不丢。写入 ES 之前一定要做数据清洗哪怕只是最基础的过滤空标题和空正文。否则 ES 里全是脏数据业务查询结果没人敢信。清洗规则根据业务需求扩展比如过滤掉发布时间在未来的数据、过滤掉重复 URL 的数据这些规则在 Spark 层用 filter 就能实现不需要单独写清洗服务。3.3 数据质量校验不能只在存储层做采集端就要卡一道文档里提到数据传输时要校验发布时间、标题、正文等解析的准确度这个校验动作很多人会放在数据入 HBase 之后做但那时候发现问题已经晚了。数据质量校验的正确姿势是在采集器解析完、推送 Kafka 之前就做一遍初筛入 HBase 之后再做一遍复核。我见过最典型的翻车现场是解析规则把发布时间字段提取错了把网页的修改时间当成了发布时间这个错误在采集端完全看不出来因为字段不为空、格式也对直到业务方发现媒体文章的时间顺序全乱了才排查出来。这种问题靠校验规则很难发现所以采集端的数据抽检很有必要。我在采集器里会加一个抽样逻辑每采集 100 条数据随机抽 1 条把解析后的数据结构和原始网页的关键信息打印出来。这不是校验规则能覆盖的而是人工确认解析逻辑仍然正确的手段。页面改版是常有的事解析规则不可能永远适配抽样检查是发现解析规则失效的最快方式。4. 任务调度与队列去重Redis 里最容易踩的三个性能陷阱采集层的核心是 Redis 缓存平台和任务调度中心这里也是性能问题最容易爆发的地方。很多人以为 Redis 够快就不会出问题但快不等于不会错尤其在队列设计和去重策略上坑非常多。4.1 任务队列用 List 还是用 Stream决定了断点续采能力任务调度中心要和 Redis 配合最常见的方式是把 URL 或任务 ID 直接塞进 List。这个方案简单直接但有一个致命缺陷如果采集器处理完任务还没来得及删除 List 里的元素就崩溃了这个任务在重启后永远不会被重新消费。用 LPOP 命令取任务取走了就没了不处理也是没了。Redis 5.0 之后的 Stream 类型就是为解决这个问题设计的。Stream 支持消费者组可以记录每个消费者的消费进度崩溃重启后可以从最后一条已消费记录继续消费。对于采集系统这种对数据完整性要求高的场景用 Stream 比用 List 稳妥得多。我这里有个教训当时系统里用 List 做任务队列某天一台采集器 OOM 崩溃重启后丢失了大约 2000 个任务这些任务对应的网页没被采集到。后来改成 Stream类似问题再没出现过。# 用 Redis Stream 做任务队列支持断点续采 XGROUP CREATE task_stream task_group 0 MKSTREAM # 采集器消费任务pending 列表里记录未确认的任务 XREADGROUP GROUP task_group consumer_1 COUNT 10 STREAMS task_stream # 处理完成后确认 XACK task_stream task_group task_id_123这套方式的关键在于 XACK。采集器从 Stream 里读到任务后任务会进入 pending 列表只有调用 XACK 确认处理完成任务才会从 pending 里移除。如果采集器崩溃pending 里的任务会一直存在重启后可以用 XPENDING 查询并重新分配。Kafka 有 offset 概念Redis Stream 的 pending 列表就相当于一个简化版的 offset。4.2 任务去重的维度URL 去重和内容指纹去重是两回事去重是采集系统中最容易出问题的环节之一。文档里提到任务处理唯一性很多人理解成按 URL 去重就够了但实际场景中往往不够。同一个新闻可能被多个网站转载标题一样、内容一样URL 完全不同。如果你只按 URL 去重一篇被转载 50 次的新闻会被采集 50 次后面清洗阶段还要花大力气做内容去重。我一般会在采集器解析完成之后做一次内容指纹去重。最简单的做法是取标题的 MD5 作为内容指纹这样同一个新闻不管 URL 是什么指纹都是一样的在存储层可以直接过滤掉重复数据。不过要注意一点如果两个网页的标题有细微差别比如一个加了【转载】前缀MD5 就会完全不同。所以指纹不能单纯依赖标题最好结合正文前 100 个字符做组合判断这样准确率会高很多。# 内容指纹去重防止转载文章重复入库 import hashlib def generate_fingerprint(title, content): # 标题和正文前100字组合生成指纹 content_prefix content[:100].strip() raw f{title.strip()}|{content_prefix} return hashlib.md5(raw.encode(utf-8)).hexdigest() # 存储前检查指纹是否存在 fp generate_fingerprint(data[title], data[content]) if redis_client.sismember(article_fingerprints, fp): return # 重复文章直接丢弃 redis_client.sadd(article_fingerprints, fp) # 定期清理 30 天前的指纹避免集合无限膨胀 redis_client.expire(article_fingerprints, 30 * 24 * 3600)指纹集合的过期策略是个容易踩的坑。如果你用的是 sadd 加上不设置过期时间这个集合会无限增长Redis 内存早晚被打满。设 30 天过期意味着一个月前的文章即使被转载也不会再采第二次不过对于绝大多数业务场景已经够用了。如果是做长期的内容追踪可以考虑把指纹存进 HBase 而不是 Redis查询效率低一点但存储空间不是问题。4.3 调度频率设置固定间隔看起来省事但容易把目标网站打挂任务调度中心负责按设置的采集频率调度采集任务这个频率怎么设很有讲究。很多人的第一反应是固定间隔比如每 30 分钟采一次。但不同信源的更新频率差别很大新闻类网站可能每几分钟就有新内容个人博客可能一个月才更新一次。统一用固定频率不是采不到数据就是浪费资源。更合理的做法是按信源类型配置不同的调度频率而且要加一个自适应调节机制。如果某个网站连续几次采集返回超时说明对方可能在限流或服务器压力大这时候要自动降低采集频率。如果某次采集发现内容更新量特别大比如出现了突发事件可以临时提高采集频率。这个逻辑在任务调度中心里做成规则就好。我在实践中会先用一段历史数据分析信源的更新规律再确定初始调度频率。新接入的信源保守一点先用 30 分钟间隔跑三天看数据量再调整。有些网站更新很规律比如每天上午九点发一批文章那调度完全可以设置为每天早上九点半采一次比每隔一小时扫一次更高效。5. 常见问题与排查调度重复、数据漏采、平台积压、解析错乱的四个实战定位这套架构跑起来之后问题不会少。我整理了几类高频问题都是生产环境实际遇到过的每个都按现象、原因、解决的顺序说清楚方便排查时直接对照。5.1 调度重复同一篇文章被采集器重复入库现象是 HBase 里出现大量标题相同但 rowkey 不同的记录。原因有两个常见来源一是任务锁没设超时时间采集器处理任务时崩溃没释放锁另一个采集器接管后重复执行二是采集器处理完任务后没有向调度中心上报完成状态调度中心等了一会儿发现没回应重新分配任务。解决方式是双管齐下。任务锁必须设置超时时间用 Redis SETNX 加 EX 参数超时时间根据任务预估执行时间设定。同时调度中心的任务状态上报要实时不能批量延迟上报否则状态更新不及时就会误判任务超时。还要提醒一句从 Kafka 消费消息处理完后offset 提交要用手动模式不要用自动提交否则数据处理完还没提交 offset 就崩溃了重启后这条消息会再被消费一遍。5.2 数据漏采某个栏目的文章总是不全现象是某个网站的分类页面下文章数量明显少于实际。原因大概率出在栏目页面只展示了最新内容没有覆盖历史文章。很多列表页是分页加载的你只采了第一页后面的全丢了。解决方式是检查栏目页是否支持分页以及链接到文章详情页的入口是否完整。还有一个隐蔽的坑有的网站列表页点击加载更多是 AJAX 异步请求URL 不变化直接抓列表页只能拿到初始加载的部分内容。这时候要分析 AJAX 接口直接采集接口返回的 JSON解析里面的 link 字段就能拿到全部文章链接。排查时先看列表页的 HTML 结构再决定是模拟点击还是调接口。5.3 Kafka 消费积压数据追不上采集速度现象是 Kafka 里 unconsume 的消息越来越多消费延迟从几分钟变成几小时。原因有两类采集器数量加多了推送速度上来了但 Spark 消费端没有做水平扩展或者消费端的某个处理环节出现了瓶颈比如 HBase 批量写入太慢拖累了整个消费链路。解决方式要先定位瓶颈在哪个环节。看 Spark UI 的 stage 耗时分布如果写入 HBase 的耗时占比高说明是存储层的问题可以增大 HBase 的批量写入大小或增加 HBase 的 region server 资源。如果是整体处理能力不够就增加 Spark 的消费者数量同时注意一个分区的数据只能被一个消费者处理要让消费者数量不超过 Kafka 分区数否则多的消费者会空闲。5.4 解析错乱正文混入导航栏或推荐链接现象是采集到的正文里有大量无关链接和广告文案。原因是解析规则太宽松正文提取时把 HTML 里的导航、页脚、推荐阅读一起抓进来了。这个问题在采集文章类网站时很常见。解决方式分两步。解析规则要更精确先用 CSS 选择器定位正文的容器节点而不是全文抓取然后提取正文时过滤掉 script、style、iframe 等标签下的内容。还要在采集器里加一个自己的校验规则比如正文的最短长度和最长长度范围超出范围就标记解析异常交给人工处理而不是直接入库。5.5 ES 索引数据与 HBase 数据不一致现象是按标题在 ES 里能查到文章但 HBase 里没有对应的完整记录。原因是 Spark 写 ES 和写 HBase 分成了两个 Stream其中一个 Stream 运行异常没有告警导致只有一边的数据。解决方式是必须给两边单独加监控指标。数据总量可以不一致但差值要在可接受范围内。另外可以在 Spark 里对每条消息做双写的时候加一个日志记录写 ES 和写 HBase 的结果如果一边失败需要重试而不是默默跳过。这个问题的本质是两条消费链路没有原子性但一般情况下不需要用事务解决监控数据差值及时发现就能把影响降到最低。6. 采集策略优化实践调度频率怎么调深度和广度怎么平衡架构搭好了、坑也排过了接下来是把采集能力调到最优的问题。这里有一个很容易被忽视的原则采集不是越快越好更不是越全越好而是在资源有限的前提下做到性价比最高。调度频率的调节逻辑我一般做成梯度策略。新信源接入后先用 30 分钟的间隔试采三天统计每天的更新数量和新内容占比。如果每天采到的有效新内容超过总量的 20%说明更新频繁可以提高到 10 分钟采一次如果三天只采到几条新内容说明更新很慢直接降到 2 小时一次省出的资源给更新快的信源。这个逻辑看起来简单但效果比固定频率好得多尤其在信源数量上百之后。# 梯度调度策略根据历史采集数据自动调整频率 def get_schedule_interval(source_id, history_data): # 统计最近7天的日均新内容数量 daily_avg calculate_avg_new_items(history_data) if daily_avg 100: return 10 # 高频信源10分钟采一次 elif daily_avg 20: return 30 # 中频信源30分钟采一次 elif daily_avg 5: return 60 # 低频信源1小时采一次 else: return 120 # 极低频信源2小时采一次用日均新内容量算出来的调度频率比拍脑袋定的间隔要靠谱得多。这里的判断阈值可以根据你的实际任务量调整但核心逻辑是一样的数据产出多的信源多投入资源产出少的信源少打扰。深度和广度的平衡也是采集策略的重头。采集需要决定爬多少层只采列表页的文章详情还是连详情页里的相关链接也一起采。我建议默认只采一层也就是列表页到详情页插入内容加粗部分为止。为什么因为每多爬一层采集时间成本和服务器压力都会翻倍而且相关链接指向的不一定是目标站点的内容。如果要扩大范围优先增加信源类型而不是增加爬取深度。从那以后我每次搭建采集系统都会强制走一遍这套流程先明确信源类型和更新规律再定调度策略采集端做初筛、存储端做复核监控系统四个维度一个不落最后再用梯度策略持续优化。这套方法帮我撑过了多次采集高峰期也希望帮到你。本文还有配套的精品资源点击获取