Scrapy使用步骤(3)

Scrapy 使用步骤(3)：Item Pipeline 管道流水线、数据清洗与持久化实战

小象实战讲义 · Python数据分析实战

在 Scrapy 架构中，Spider 的职责是负责“抓”和“析”，而真正负责“洗”和“存”的核心组件则是 Item Pipeline（数据管道）。当 Spider 的解析函数通过 yield item 产出一个结构化数据对象后，引擎会自动将该 Item 送入配置好的 Pipeline 流水线中。在这里，数据将经历严格的字段清洗、缺失值校验、业务规则去重，并最终被格式化写入 CSV 文件、JSON 文件或关系型数据库中。本节我们将深入学习 Item Pipeline 的生命周期、权重调度机制与多通道持久化实现。

💡 核心导读

• Item Pipeline 的四大核心业务场景：
  • 数据清洗（去除杂质文本、单位换算、类型强制转换）；
  • 数据完整性验证与异常丢弃（DropItem 机制）；
  • 数据指纹去重（防止重复数据入库）；
  • 持久化落地（保存为 CSV / JSON / SQLite / MySQL 等）。
• Pipeline 三大生命周期方法：
  • open_spider(self, spider)：爬虫启动初始化；
  • process_item(self, item, spider)：核心加工（必须 return item 或 raise DropItem）；
  • close_spider(self, spider)：爬虫结束资源释放。
• 管道优先级权重（0~1000）：数值越小优先级越高。

1. Item Pipeline 生命周期与流水线编排

                    ┌─────────────────────────────────────────┐
                    │      Spider 解析产出 (yield Item)       │
                    └────────────────────┬────────────────────┘
                                         │
                                         ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 1. CleanPipeline (权重 300) : 字段清洗与类型转换                            │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │ return item
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 2. CsvExportPipeline (权重 500) : 格式化写入本地 CSV 文件                   │
└─────────────────────────────────────────────────────────────────────────────┘

2. Python 代码实战：编写数据清洗与 CSV 持久化 Pipeline

# 示例 1：定义清洗管道与 CSV 持久化管道

import csv
import os
from scrapy.exceptions import DropItem

class AqiCleanPipeline:
    """负责数据清洗与合法性校验"""
    def process_item(self, item, spider):
        if not item.get("city_name"):
            raise DropItem("❌ 缺失城市名称，丢弃记录！")
            
        item["city_name"] = str(item["city_name"]).strip()
        try:
            aqi_val = int(item.get("aqi", 0))
            if aqi_val < 0 or aqi_val > 1000:
                raise ValueError("AQI 范围异常")
            item["aqi"] = aqi_val
        except Exception:
            raise DropItem(f"❌ 城市【{item['city_name']}】AQI 异常")
            
        return item  # 必须显式 return item

class CsvExportPipeline:
    """负责流式写入 CSV"""
    def __init__(self):
        self.filename = "cities_aqi.csv"
        self.file = None
        self.writer = None
        
    def open_spider(self, spider):
        self.file = open(self.filename, "w", newline="", encoding="utf-8-sig")
        self.writer = csv.DictWriter(self.file, fieldnames=["city_name", "aqi", "quality_level"])
        self.writer.writeheader()

    def process_item(self, item, spider):
        self.writer.writerow({
            "city_name": item.get("city_name"),
            "aqi": item.get("aqi"),
            "quality_level": item.get("quality_level")
        })
        return item

    def close_spider(self, spider):
        if self.file:
            self.file.close()

print("✅ Pipeline 类定义完成！")

接下来模拟流水线串联执行：

# 示例 2：模拟 Pipeline 链式处理

pipelines = [AqiCleanPipeline(), CsvExportPipeline()]
mock_spider = type("Spider", (), {"name": "test"})()

# 1. 触发 open_spider
for p in pipelines:
    if hasattr(p, "open_spider"):
        p.open_spider(mock_spider)

# 2. 模拟 Item 流转
test_data = [
    {"city_name": " 北京 ", "aqi": "35", "quality_level": "优"},
    {"city_name": "", "aqi": "50"},  # 无城市名，将被丢弃
    {"city_name": "上海", "aqi": "55", "quality_level": "良"}
]

for raw in test_data:
    item = raw
    try:
        for p in pipelines:
            item = p.process_item(item, mock_spider)
        print("✅ 成功流转并入库:", item)
    except DropItem as e:
        print("⚠️ 拦截异常数据:", e)

# 3. 触发 close_spider
for p in pipelines:
    if hasattr(p, "close_spider"):
        p.close_spider(mock_spider)

if os.path.exists("cities_aqi.csv"):
    os.remove("cities_aqi.csv")

📝 动手练一练

1. 机制思考题：在 process_item(self, item, spider) 方法中，如果开发者忘记写 return item，会导致什么后果？

   👉 点击查看参考答案

   参考答案： 函数会默认返回 None。排在后面的所有后续 Pipeline（如写入数据库或 CSV 的管道）接收到的输入都将是 None，导致后续管道无法处理数据甚至崩溃。因此必须显式 return item。

2. 配置题：在 settings.py 中如何配置使得 CleanPipe 先于 DbPipe 执行？

   👉 点击查看参考答案

   参考答案：

   ITEM_PIPELINES = {
       'aqi_proj.pipelines.CleanPipe': 300,  # 权重小先执行
       'aqi_proj.pipelines.DbPipe': 500,     # 权重大后执行
   }

本章小结

在本节中，我们全面掌握了 Scrapy 数据加工与落盘的核心载体——Item Pipeline：

• 深刻理解了 Pipeline 在数据清洗、验证丢弃（DropItem）与持久化中的分工；
• 掌握了 open_spider、process_item 与 close_spider 三大生命周期方法；
• 掌握了 settings.py 中通过权重整数编排多级流水线的工程实践。

📋 行动清单

• 牢记 process_item 必须 return item 的规则。
• 做好准备，进入下一小节学习《Scrapy 使用步骤(4)：命令行执行与配置优化》！

—— 小象教研组