Scrapy使用步骤(4)

Scrapy Pipeline：数据清洗与持久化

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

在上一节中，我们学习了如何使用 Scrapy 的 Spider 来解析网页并提取出结构化的数据（Item）。但解析出的数据往往不是最终形态，我们通常需要对其进行清洗（如去重、填充缺失值），并最终保存到本地文件或数据库中。本节将讲解 Scrapy 框架中的 Pipeline 组件，它就像一个管道，负责对 Spider 爬取到的 Item 进行一系列的后处理，并最终完成数据的持久化。

💡 核心导读

• Pipeline 的作用：理解 Pipeline 作为“数据处理管道”的核心思想，它负责接收、处理和输出 Spider 产生的 Item。
• Pipeline 的配置：掌握如何在 settings.py 中启用和配置自定义的 Pipeline。
• Pipeline 的编写：学习编写一个自定义的 Pipeline 类，实现数据的清洗、转换和保存（以 CSV 文件为例）。
• 实战演练：完成一个完整的 Scrapy 项目，将爬取的城市名称和链接数据，经过 Pipeline 处理后，保存为 CSV 文件。

什么是 Pipeline？

在 Scrapy 框架中，Pipeline 是一个专门用于处理 Spider 爬取到的 Item 的组件。你可以把它想象成一个工厂的流水线：

1. Spider 是原料采集车间，负责从网络上“采集”原始数据（Item）。
2. Pipeline 是加工车间，负责对采集到的原料（Item）进行清洗、筛选、组装等处理。
3. 最终，处理好的成品（数据）被包装、入库（保存到文件或数据库）。

这种设计将数据爬取（Spider）与数据处理（Pipeline）解耦，使得代码结构更清晰，也便于维护和扩展。例如，你可以在 Pipeline 中实现：

• 数据清洗：去除重复项、过滤无效数据、填充缺失值。
• 数据验证：检查数据格式是否符合预期。
• 数据持久化：将数据保存为 CSV、JSON、XML 文件，或写入 MySQL、MongoDB 等数据库。

配置 Pipeline

Pipeline 的启用和配置主要在项目根目录下的 settings.py 文件中进行。

1. 找到配置项：打开 settings.py 文件，找到名为 ITEM_PIPELINES 的配置字典。它默认是被注释掉的。
2. 启用 Pipeline：取消该字典的注释。字典的键（Key）是你将要编写的 Pipeline 类的完整导入路径，值（Value）是一个代表该 Pipeline 执行优先级的整数（范围 0-1000，数值越小优先级越高）。当有多个 Pipeline 时，Scrapy 会根据优先级决定执行顺序。

# settings.py 中的配置示例
ITEM_PIPELINES = {
    # ‘你的项目名.pipelines.你的Pipeline类名’ : 优先级,
    'myproject.pipelines.CityItemPipeline': 300,
}

优先级说明：优先级数字越小，越早执行。通常使用默认值 300 即可。如果你有多个 Pipeline，例如一个用于数据清洗（优先级 200），一个用于数据保存（优先级 400），那么清洗 Pipeline 会先执行。

编写自定义 Pipeline

Pipeline 本质上是一个 Python 类。我们将在 pipelines.py 文件中创建它。这个类通常需要实现三个核心方法：

• open_spider(self, spider): 当 Spider 被打开时调用。通常在这里进行初始化工作，如打开文件、连接数据库。
• process_item(self, item, spider): 这是 Pipeline 的核心。每个被 Spider 爬取到的 Item 都会经过这个方法。你可以在这里对 Item 进行处理，并必须返回一个 Item 对象（或抛出 DropItem 异常丢弃该 Item）。
• close_spider(self, spider): 当 Spider 被关闭时调用。通常在这里进行收尾工作，如关闭文件、断开数据库连接。

下面，我们以实现一个将 Item 保存为 CSV 文件的 CityItemPipeline 为例。请注意：由于本课程环境限制（禁用了 Scrapy 库），以下代码为原理性伪代码，用于说明 Pipeline 的编写逻辑。在实际的 Scrapy 项目中，你可以直接使用 scrapy.exporters 模块中的 CsvItemExporter 来简化导出工作。

# 以下为 Scrapy Pipeline 原理性伪代码，展示核心结构
# 实际 Scrapy 项目请使用 scrapy.exporters.CsvItemExporter

class CityItemPipeline:
    def open_spider(self, spider):
        """Spider 启动时执行，初始化文件等资源"""
        # 伪代码：打开一个 CSV 文件准备写入
        # self.file = open('cities.csv', 'w', newline='', encoding='utf-8')
        # self.writer = csv.DictWriter(self.file, fieldnames=['city_name', 'city_link'])
        # self.writer.writeheader()
        print("Pipeline: 开始处理数据，准备写入文件...")

    def process_item(self, item, spider):
        """处理每一个 Item"""
        # 伪代码：这里可以进行数据清洗，例如去除空值
        # if item.get('city_name') and item.get('city_link'):
        #     self.writer.writerow({
        #         'city_name': item['city_name'],
        #         'city_link': item['city_link']
        #     })
        # else:
        #     print(f"丢弃无效数据: {item}")
        print(f"Pipeline: 正在处理 Item -> {dict(item)}")
        # 必须返回 item，以供后续可能的 Pipeline 处理
        return item

    def close_spider(self, spider):
        """Spider 关闭时执行，释放资源"""
        # 伪代码：关闭文件
        # self.file.close()
        print("Pipeline: 数据处理完成，文件已保存。")

实战：模拟 Pipeline 数据处理流程

虽然我们无法在讲义中运行真实的 Scrapy 项目，但我们可以用纯 Python 模拟 Pipeline 的核心思想：接收数据 -> 处理数据 -> 输出数据。假设我们已经通过某种方式（比如上一节的解析逻辑）获得了一个城市数据列表，现在需要对其进行清洗并保存。

下面的代码块完全使用 Python 标准库，模拟了 Pipeline 的流程：

import csv
from io import StringIO

# 模拟从 Spider 解析得到的原始数据列表 (Item 列表)
raw_city_items = [
    {'city_name': '北京', 'city_link': 'https://example.com/beijing'},
    {'city_name': '上海', 'city_link': 'https://example.com/shanghai'},
    {'city_name': '', 'city_link': 'https://example.com/empty'},  # 无效数据：城市名为空
    {'city_name': '广州', 'city_link': 'https://example.com/guangzhou'},
    {'city_name': '深圳', 'city_link': 'https://example.com/shenzhen'},
    {'city_name': '广州', 'city_link': 'https://example.com/guangzhou'},  # 重复数据
]

# 1. 数据清洗函数 (模拟 Pipeline 的 process_item 逻辑)
def clean_city_item(item):
    """清洗单个城市数据项"""
    # 规则1: 城市名不能为空
    if not item.get('city_name'):
        print(f"清洗: 丢弃城市名为空的数据 -> {item}")
        return None
    # 规则2: 链接不能为空（这里假设链接都有值）
    if not item.get('city_link'):
        print(f"清洗: 丢弃链接为空的数据 -> {item}")
        return None
    # 其他清洗规则可以在此添加...
    return item

# 2. 主处理流程 (模拟 Spider -> Pipeline 的数据流)
cleaned_items = []
seen = set()  # 用于去重的集合

for raw_item in raw_city_items:
    cleaned_item = clean_city_item(raw_item)
    
    if cleaned_item is None:
        continue  # 丢弃无效数据
    
    # 规则3: 基于城市名和链接进行去重
    item_key = (cleaned_item['city_name'], cleaned_item['city_link'])
    if item_key in seen:
        print(f"去重: 丢弃重复数据 -> {cleaned_item}")
        continue
    
    seen.add(item_key)
    cleaned_items.append(cleaned_item)

print(f"清洗和去重后，有效数据项数量: {len(cleaned_items)}")

# 3. 数据持久化 (模拟保存为 CSV)
# 使用 StringIO 模拟文件写入，避免实际生成文件
output = StringIO()
fieldnames = ['city_name', 'city_link']
writer = csv.DictWriter(output, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(cleaned_items)

# 获取模拟的 CSV 文件内容
csv_content = output.getvalue()
output.close()

print("\n--- 生成的 CSV 文件内容 ---")
print(csv_content)

清洗: 丢弃城市名为空的数据 -> {'city_name': '', 'city_link': 'https://example.com/empty'}
去重: 丢弃重复数据 -> {'city_name': '广州', 'city_link': 'https://example.com/guangzhou'}
清洗和去重后，有效数据项数量: 4

--- 生成的 CSV 文件内容 ---
city_name,city_link
北京,https://example.com/beijing
上海,https://example.com/shanghai
广州,https://example.com/guangzhou
深圳,https://example.com/shenzhen

通过上面的模拟，我们清晰地演示了 Pipeline 的核心价值：在数据被最终保存前，提供了一个集中、可配置的环节来进行各种数据质量治理工作。

关于 Scrapy 内置导出器的补充说明

在真实的 Scrapy 项目中，为了简化将 Item 导出为各种格式文件的过程，Scrapy 提供了 scrapy.exporters 模块。例如，使用 CsvItemExporter 可以非常方便地将 Item 导出为 CSV 文件。

需要注意的一个常见问题：某些早期版本或配置下的 CsvItemExporter 在导出 CSV 时，可能会在每行数据之间插入一个空行。如果遇到这个问题，你有两种解决思路：

1. 后续处理：在拿到 CSV 文件后，使用我们后面将学到的 Pandas 进行读取和清洗，Pandas 可以轻松处理多余的空行。
2. 自定义导出逻辑：你可以完全自己编写文件写入逻辑，不使用内置的 Exporter，从而精确控制输出格式。

📝 动手练一练

1. 概念辨析：Scrapy 框架中，Spider 和 Pipeline 的主要分工是什么？如果我想在数据保存到数据库前，先过滤掉所有“价格”字段低于 10 元的商品记录，这个逻辑应该放在哪个组件中实现？

2. 代码实践：参考上面的模拟 Pipeline 代码，假设原始数据中增加了“空气质量指数(AQI)”字段。请修改清洗函数 clean_city_item，增加一条规则：如果 AQI 值大于 150（表示污染），则将该条数据也丢弃。原始数据如下：

   raw_city_items = [
       {'city_name': '北京', 'city_link': '...', 'aqi': 120},
       {'city_name': '石家庄', 'city_link': '...', 'aqi': 180},
       {'city_name': '厦门', 'city_link': '...', 'aqi': 45},
   ]

def clean_city_item_v2(item):
    """清洗单个城市数据项，增加 AQI 过滤"""
    # 规则1: 城市名不能为空
    if not item.get('city_name'):
        print(f"清洗: 丢弃城市名为空的数据 -> {item}")
        return None
    # 规则2: 链接不能为空
    if not item.get('city_link'):
        print(f"清洗: 丢弃链接为空的数据 -> {item}")
        return None
    # 新增规则3: AQI 不能大于150
    if item.get('aqi', 0) > 150:  # 使用 get 防止 aqi 字段缺失，默认值为0
        print(f"清洗: 丢弃 AQI > 150 的数据 -> {item}")
        return None
    return item

# 测试新函数
test_items = [
    {'city_name': '北京', 'city_link': '...', 'aqi': 120},
    {'city_name': '石家庄', 'city_link': '...', 'aqi': 180},
    {'city_name': '厦门', 'city_link': '...', 'aqi': 45},
]
for it in test_items:
    result = clean_city_item_v2(it)
    print(f"输入: {it} -> 输出: {result}")

输入: {'city_name': '北京', 'city_link': '...', 'aqi': 120} -> 输出: {'city_name': '北京', 'city_link': '...', 'aqi': 120}
清洗: 丢弃 AQI > 150 的数据 -> {'city_name': '石家庄', 'city_link': '...', 'aqi': 180}
输入: {'city_name': '石家庄', 'city_link': '...', 'aqi': 180} -> 输出: None
输入: {'city_name': '厦门', 'city_link': '...', 'aqi': 45} -> 输出: {'city_name': '厦门', 'city_link': '...', 'aqi': 45}

本章小结

本节我们完成了 Scrapy 框架核心流程的最后一块拼图——Pipeline。我们了解到，Pipeline 是 Scrapy 项目中负责数据后处理的“加工车间”，它将 Spider 的爬取逻辑与数据的清洗、持久化逻辑优雅地分离。

📋 行动清单

• 理解 Pipeline 机制：在脑海中建立起“Spider 生产 Item -> Pipeline 加工 Item -> 输出最终结果”的数据流图。
• 掌握配置与编写：记住在 settings.py 中配置 ITEM_PIPELINES 的方法，以及自定义 Pipeline 类需要实现的三个核心方法（open_spider, process_item, close_spider）。
• 模拟数据流：尝试用纯 Python 代码模拟一个简单的数据处理管道，对一组字典列表进行清洗（去空、去重）并输出，巩固“数据流”处理的思想。这种思想在后续学习 Pandas 等数据处理库时同样适用。

至此，关于 Scrapy 框架的核心使用步骤我们就介绍完毕了。从创建项目、定义数据结构、编写爬虫主体到配置处理管道，你已经掌握了一个现代爬虫框架的基本工作流程。虽然在实际项目中可能会遇到更复杂的页面结构和反爬策略，但这套基础方法论是解决问题的根本。

—— 小象教研组