海量小说数据采集：Spark 爬虫系统设计

原创

小白学大数据

发布于 2025-12-30 16:52:19

1820

在数字阅读产业高速发展的背景下，海量小说数据成为内容分析、用户画像构建、版权监测等业务的核心资产。传统单机爬虫面对百万级甚至亿级小说资源时，存在采集效率低、任务调度难、数据处理能力弱等问题。Apache Spark 作为分布式计算框架，凭借其内存计算、弹性分布式数据集（RDD）和分布式任务调度能力，成为构建海量小说数据采集系统的理想选择。本文将从系统架构、核心模块设计、技术实现等维度，详解基于 Spark 的小说数据爬虫系统构建过程。

一、系统设计核心目标与架构

1.1 核心目标

针对小说数据采集的特殊性（多源站点、内容分散、反爬机制多样、数据量大），Spark 爬虫系统需实现以下目标：

分布式采集：支持百台级节点并行爬取，单日采集能力达千万级小说章节；
容错性：节点故障时任务自动重试、重新分配，避免数据丢失；
数据清洗：实时处理乱码、格式不统一、广告冗余内容；
可扩展性：适配不同小说站点的反爬策略，支持新增站点快速接入；
数据落地：结构化存储采集结果，支持后续分析与检索。

1.2 系统整体架构

Spark 爬虫系统采用分层设计，从上至下分为：

任务调度层：基于 Spark Standalone/YARN 实现分布式任务分发，管理爬取队列与节点资源；
爬虫核心层：封装分布式爬取逻辑，包含请求发送、反爬突破、内容解析模块；
数据处理层：通过 Spark RDD/DataFrame 完成数据清洗、去重、结构化转换；
存储层：分布式文件系统（HDFS）存储原始数据，HBase/MySQL 存储结构化数据。

二、核心模块设计与技术实现

2.1 分布式任务调度模块

Spark 的核心优势在于将爬取任务拆分为多个 Partition，分发至不同 Executor 节点并行执行。系统首先构建小说站点的 URL 种子池，通过 Spark 的parallelize方法将 URL 列表转化为 RDD，每个 Partition 对应一批待爬取的 URL，由不同节点处理。

关键设计点：

任务分片策略：按站点域名、小说分类进行分片，避免单站点爬取频率过高触发反爬；
任务优先级：热门小说、新上架小说的 URL 分配更高优先级，优先执行；
失败重试机制：基于 Spark 的retry算子和 Checkpoint 机制，对失败任务标记并重新调度。

2.2 反爬突破模块

小说站点常见反爬手段包括 IP 封禁、User-Agent 检测、Cookie 验证、动态页面渲染，系统针对性设计解决方案：

IP 代理池：整合分布式代理 IP 资源，推荐使用亿牛云隧道代理
动态请求头：随机生成 User-Agent、Referer，模拟真实浏览器请求；
动态页面处理：集成 Selenium+ChromeDriver，针对 JS 渲染的小说章节页面，通过 Spark Executor 节点本地启动浏览器进程渲染页面。

2.3 数据解析与清洗模块

爬取的小说数据包含 HTML 源码、冗余广告、乱码等问题，需通过 Spark 分布式计算完成清洗：

结构化解析：使用 Jsoup 解析 HTML，提取小说标题、作者、章节内容、更新时间等核心字段；
数据清洗：通过正则表达式剔除广告、无关链接，统一字符编码（UTF-8），处理空值和重复数据；
质量校验：设置数据校验规则（如章节内容长度、标题非空），过滤无效数据。

2.4 数据存储模块

原始数据：爬取的 HTML 源码存储至 HDFS，路径按 “站点 / 日期 / 小说 ID” 划分，便于回溯；
结构化数据：清洗后的小说信息通过 Spark DataFrame 写入 HBase（适合海量非结构化数据）或 MySQL（适合高频查询的核心数据）；
元数据管理：记录爬取状态、节点负载、反爬触发次数，存储至 Redis 便于实时查询。

三、代码实现过程

3.1 环境准备

Spark 3.3.0（集群模式）；
JDK 1.8；
依赖库：Jsoup（HTML 解析）、HttpClient（请求发送）、HBase Client（数据存储）、Selenium（动态页面渲染）。

3.2 核心代码实现

步骤 1：初始化 Spark 上下文

scala

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

object NovelCrawler {
  def main(args: Array[String]): Unit = {
    // Spark配置：设置应用名、集群模式
    val conf = new SparkConf()
      .setAppName("NovelCrawler_Spark")
      .setMaster("yarn") // 集群模式使用yarn，本地测试用local[*]
      .set("spark.executor.memory", "4g") // 每个Executor内存
      .set("spark.cores.max", "100") // 最大使用核数
      .set("spark.task.maxFailures", "3") // 任务最大失败次数
    
    val sc = new SparkContext(conf)
    sc.setCheckpointDir("hdfs://cluster/checkpoint/novel_crawler") // 容错检查点
    
    // 加载URL种子池（从HDFS读取待爬取的小说URL列表）
    val urlRDD: RDD[String] = sc.textFile("hdfs://cluster/novel/seed_urls.txt")
    
    // 分布式爬取
    val novelDataRDD: RDD[NovelInfo] = urlRDD.mapPartitions(urls => {
      // 每个Partition初始化爬虫客户端（避免重复创建连接）
      val crawler = new NovelCrawlerClient()
      urls.map(url => crawler.crawl(url))
    }).filter(_ != null) // 过滤爬取失败的数据
    
    // 数据清洗
    val cleanedNovelRDD: RDD[NovelInfo] = novelDataRDD.map(novel => {
      // 清洗章节内容：剔除广告、乱码
      val cleanContent = NovelCleaner.cleanContent(novel.content)
      // 统一字符编码
      val normalizedTitle = new String(novel.title.getBytes("ISO-8859-1"), "UTF-8")
      NovelInfo(novel.id, normalizedTitle, novel.author, cleanContent, novel.updateTime)
    })
    
    // 数据落地：写入HBase
    cleanedNovelRDD.foreachPartition(novels => {
      val hbaseClient = new HBaseClient("novel_table") // 初始化HBase客户端
      novels.foreach(novel => {
        hbaseClient.put(
          rowKey = novel.id,
          family = "content",
          columns = Map(
            "title" -> novel.title,
            "author" -> novel.author,
            "content" -> novel.content,
            "update_time" -> novel.updateTime.toString
          )
        )
      })
      hbaseClient.close()
    })
    
    sc.stop()
  }
}

步骤 2：爬虫客户端实现（NovelCrawlerClient）scala

import org.apache.http.client.methods.HttpGet
import org.apache.http.impl.client.CloseableHttpClient
import org.apache.http.impl.client.HttpClients
import org.jsoup.Jsoup
import org.jsoup.nodes.Document
import org.openqa.selenium.WebDriver
import org.openqa.selenium.chrome.ChromeDriver
import java.util.concurrent.TimeUnit

// 小说信息样例类
case class NovelInfo(
  id: String, // 小说ID（URL哈希生成）
  title: String, // 标题
  author: String, // 作者
  content: String, // 章节内容
  updateTime: Long // 更新时间戳
)

class NovelCrawlerClient {
  // 初始化HTTP客户端
  private val httpClient: CloseableHttpClient = HttpClients.createDefault()
  // 初始化Selenium WebDriver（处理动态页面）
  private val driver: WebDriver = {
    System.setProperty("webdriver.chrome.driver", "/opt/chromedriver") // 集群节点ChromeDriver路径
    val d = new ChromeDriver()
    d.manage().timeouts().pageLoadTimeout(30, TimeUnit.SECONDS)
    d
  }
  
  // 核心爬取方法
  def crawl(url: String): NovelInfo = {
    try {
      // 1. 发送请求（带代理、请求头）
      val request = new HttpGet(url)
      request.addHeader("User-Agent", getRandomUserAgent())
      request.addHeader("Referer", "https://www.novel-site.com")
      // 设置代理IP（从代理池获取）
      val proxy = ProxyPool.getRandomProxy()
      if (proxy != null) {
        val httpHost = new HttpHost(proxy.ip, proxy.port)
        request.setConfig(RequestConfig.custom().setProxy(httpHost).build())
      }
      
      // 2. 响应处理
      val response = httpClient.execute(request)
      val statusCode = response.getStatusLine().getStatusCode()
      
      if (statusCode == 200) {
        // 3. 解析页面（区分静态/动态页面）
        val document: Document = if (isDynamicPage(url)) {
          // 动态页面使用Selenium渲染
          driver.get(url)
          Jsoup.parse(driver.getPageSource())
        } else {
          // 静态页面直接解析
          val entity = response.getEntity()
          Jsoup.parse(EntityUtils.toString(entity), "UTF-8")
        }
        
        // 4. 提取核心信息
        val title = document.select("h1.novel-title").text()
        val author = document.select("span.author").text()
        val content = document.select("div.chapter-content").text()
        val updateTime = document.select("span.update-time").attr("data-time").toLong
        val id = MD5Util.md5(url) // 生成唯一ID
        
        NovelInfo(id, title, author, content, updateTime)
      } else {
        // 非200状态码标记失败
        null
      }
    } catch {
      case e: Exception =>
        e.printStackTrace()
        null
    }
  }
  
  // 随机生成User-Agent
  private def getRandomUserAgent(): String = {
    val userAgents = Array(
      "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
      "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
      "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/114.0"
    )
    userAgents(scala.util.Random.nextInt(userAgents.length))
  }
  
  // 判断是否为动态页面（根据URL特征）
  private def isDynamicPage(url: String): Boolean = {
    url.contains("dynamic_chapter") || url.contains("js_render")
  }
  
  // 关闭资源
  def close(): Unit = {
    httpClient.close()
    driver.quit()
  }
}

步骤 3：数据清洗工具类（NovelCleaner）scala

import java.util.regex.Pattern

object NovelCleaner {
  // 清洗章节内容：剔除广告、无关链接、特殊字符
  def cleanContent(content: String): String = {
    if (content == null || content.isEmpty) return ""
    
    // 正则剔除广告文本
    val adPattern = Pattern.compile("【.*?广告.*?】|点击阅读.*?|VIP章节.*?|\\s{4,}")
    val adMatcher = adPattern.matcher(content)
    var cleanContent = adMatcher.replaceAll("")
    
    // 剔除特殊字符和乱码
    cleanContent = cleanContent.replaceAll("[^\\u4e00-\\u9fa5a-zA-Z0-9，。！？；：、\\s]", "")
    
    // 去除多余空格和换行
    cleanContent = cleanContent.replaceAll("\\n+", "\n").replaceAll("\\s+", " ")
    
    cleanContent
  }
}

步骤 4：HBase 客户端实现（HBaseClient）

scala

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Connection
import org.apache.hadoop.hbase.client.ConnectionFactory
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.client.Table
import org.apache.hadoop.hbase.util.Bytes

class HBaseClient(tableName: String) {
  private val conf = HBaseConfiguration.create()
  conf.set("hbase.zookeeper.quorum", "node1,node2,node3") // Zookeeper地址
  private val connection: Connection = ConnectionFactory.createConnection(conf)
  private val table: Table = connection.getTable(TableName.valueOf(tableName))
  
  // 写入数据到HBase
  def put(rowKey: String, family: String, columns: Map[String, String]): Unit = {
    val put = new Put(Bytes.toBytes(rowKey))
    columns.foreach { case (col, value) =>
      put.addColumn(
        Bytes.toBytes(family),
        Bytes.toBytes(col),
        Bytes.toBytes(value)
      )
    }
    table.put(put)
  }
  
  // 关闭连接
  def close(): Unit = {
    table.close()
    connection.close()
  }
}

四、系统优化与性能调优

4.1 性能优化策略

内存计算优化：将高频访问的 URL 种子池、代理 IP 池通过 Spark 广播变量（Broadcast）分发至各节点，避免重复传输；
数据本地化：将待爬取的 URL 按节点所在区域分片，减少跨节点数据传输；
批处理优化：调整spark.sql.shuffle.partitions参数（默认 200），根据数据量设置合理的分区数，避免小文件过多；
反爬适配：针对不同站点调整爬取频率，通过 Spark 累加器（Accumulator）统计各站点的反爬触发次数，动态调整爬取策略。

4.2 容错与监控

任务重试：通过spark.task.maxFailures设置任务最大失败次数，失败任务自动重新调度；
监控指标：集成 Prometheus+Grafana 监控 Spark 集群的 CPU、内存使用率，爬取成功率，数据清洗通过率等核心指标；
日志管理：将各节点的爬取日志输出至 ELK 平台，便于定位反爬触发、节点故障等问题。

五、总结

基于 Spark 的海量小说数据爬虫系统，通过分布式计算解决了传统单机爬虫的效率瓶颈，同时结合反爬突破、数据清洗、分布式存储等能力，实现了海量小说数据的高效、稳定采集。该系统可适配不同小说站点的特征，通过灵活的参数调优和模块扩展，满足数字阅读行业对大规模数据采集的需求。在实际应用中，需结合站点反爬策略的变化持续优化爬取逻辑，并通过监控体系保障系统的稳定性，最终为小说内容分析、版权保护等业务提供可靠的数据支撑。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python3

python爬虫

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python3

python爬虫

登录后参与评论

0 条评论

热度