2.3 Item Pipelines (数据管道)

2.3 Item Pipelines (数据管道)

Scrapy Item Pipelines (数据管道) 详解

在 Scrapy 框架中，Item Pipelines 扮演着数据处理的中枢角色。当 Spider 成功提取数据（Items）后，这些数据并不会直接被存储，而是会流经 Item Pipelines 进行一系列的处理，如数据清洗、验证、去重、存储等。Item Pipelines 提供了一种可配置且可扩展的机制，用于管理和转换爬取的数据。

1. Item Pipelines 的作用

Item Pipelines 主要负责以下任务：

• 数据清洗 (Cleaning): 清理 HTML 数据，移除不必要的空格、标签或其他不需要的内容。

• 数据验证 (Validation): 检查 Item 中是否存在必填字段，字段类型是否正确，数据格式是否符合要求等。

• 数据去重 (Duplication Removal): 识别并移除重复的 Item，确保数据的唯一性。

• 数据存储 (Storage): 将处理后的数据存储到数据库（如 MySQL、MongoDB）、文件（如 CSV、JSON）或其他存储介质中。

• 图片/文件下载 (File/Image Download): 下载 Item 中包含的图片或文件，并存储到本地或云存储服务。

2. Item Pipeline 的结构

一个 Item Pipeline 是一个独立的 Python 类，它必须实现以下方法（可选）：

• process_item(self, item, spider): 这是每个 Item Pipeline 必须实现的方法。它接收一个 Item 对象和一个 Spider 对象作为参数，并返回一个 Item 对象（可以是原始 Item 或修改后的 Item）或抛出一个 DropItem 异常。

• open_spider(self, spider): 当 Spider 开启时，该方法被调用。可以在这里进行一些初始化操作，如连接数据库。

• close_spider(self, spider): 当 Spider 关闭时，该方法被调用。可以在这里进行一些清理操作，如关闭数据库连接。

• from_crawler(cls, crawler): 这是一个类方法，用于从 Crawler 对象中获取配置信息。这通常用于创建 Item Pipeline 实例。

3. Item Pipeline 的实现

下面通过几个示例来演示如何实现 Item Pipelines。

3.1 数据验证 Pipeline

这个 Pipeline 用于验证 Item 中是否存在必填字段，如果不存在则丢弃该 Item。

from scrapy.exceptions import DropItem
class PricePipeline:
    vat_factor = 1.15
    def process_item(self, item, spider):
        if item['price']:
            if item['price_excludes_vat']:
                item['price'] = item['price'] * self.vat_factor
            return item
        else:
            return DropItem("Missing price in %s" % item)
class DuplicatesPipeline:
    def __init__(self):
        self.ids_seen = set()
    def process_item(self, item, spider):
        if item['id'] in self.ids_seen:
            raise DropItem("Duplicate item found: %s" % item)
        else:
            self.ids_seen.add(item['id'])
            return item
​

3.2 数据存储 Pipeline

这个 Pipeline 用于将 Item 存储到 JSON 文件中。

import json
class JsonWriterPipeline:
    def open_spider(self, spider):
        self.file = open('items.json', 'w')
    def close_spider(self, spider):
        self.file.close()
    def process_item(self, item, spider):
        line = json.dumps(dict(item)) + "\n"
        self.file.write(line)
        return item
​

3.3 图片下载 Pipeline

Scrapy 提供了内置的 ImagesPipeline 用于下载图片。要使用它，需要在 settings.py 中配置 ITEM_PIPELINES 和 IMAGES_STORE。

# settings.py
ITEM_PIPELINES = {
    'myproject.pipelines.MyImagesPipeline': 1,
}
IMAGES_STORE = '/path/to/your/images/directory' #本地存储路径
#IMAGES_STORE = 's3://yourbucket/images' #s3存储路径
#可选配置
#IMAGES_EXPIRES = 90  #图片过期时间（天）
#IMAGES_THUMBS = {
#    'small': (50, 50),
#    'big': (600, 600),
#}  #生成缩略图
#IMAGES_MIN_HEIGHT = 110  #图片最小高度
#IMAGES_MIN_WIDTH = 110  #图片最小宽度
​

然后，创建一个自定义的 ImagesPipeline 类，并重写 file_path 方法来定义图片存储路径。

from scrapy.pipelines.images import ImagesPipeline
import scrapy
class MyImagesPipeline(ImagesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        image_guid = request.url.split('/')[-1]
        return 'full/%s' % (image_guid)
    def item_completed(self, results, item, info):
        image_paths = [x['path'] for ok, x in results if ok]
        if not image_paths:
            raise DropItem("Item contains no images")
        item['image_paths'] = image_paths
        return item
    def get_media_requests(self, item, info):
        for image_url in item['image_urls']:
            yield scrapy.Request(image_url)
​

4. Item Pipeline 的配置

要启用 Item Pipelines，需要在 settings.py 文件中配置 ITEM_PIPELINES。ITEM_PIPELINES 是一个字典，键是 Pipeline 类的路径，值是 Pipeline 的优先级。优先级越低，Pipeline 越先被执行。

ITEM_PIPELINES = {
    'myproject.pipelines.PricePipeline': 300,
    'myproject.pipelines.JsonWriterPipeline': 800,
    'myproject.pipelines.MyImagesPipeline': 1,
}
​

5. Item Pipeline 的使用流程

1. 定义 Item: 首先，需要在 items.py 文件中定义 Item 类，用于存储爬取的数据。

2. 编写 Spider: 在 Spider 中，提取数据并填充 Item 对象。

3. 返回 Item: Spider 将填充后的 Item 对象返回。

4. Pipeline 处理: Item 对象依次通过 ITEM_PIPELINES 中配置的 Pipeline 进行处理。

5. 数据存储: 最终，处理后的数据被存储到指定的存储介质中。

6. 最佳实践

• 模块化: 将不同的数据处理逻辑拆分成独立的 Pipeline，提高代码的可维护性和可重用性。

• 配置化: 将 Pipeline 的配置信息（如数据库连接信息）放在 settings.py 文件中，方便修改和管理。

• 异常处理: 在 Pipeline 中进行适当的异常处理，避免程序崩溃。

• 日志记录: 使用 Scrapy 的日志系统记录 Pipeline 的运行状态，方便调试和排错。

• 性能优化: 对于大规模数据处理，可以考虑使用异步或多线程来提高 Pipeline 的处理速度。

7. 高级用法

• 动态 Pipeline: 根据 Spider 的类型或配置，动态地启用或禁用某些 Pipeline。

• 自定义 ItemLoader: 使用 ItemLoader 可以更方便地进行数据清洗和转换。

• 与第三方库集成: Item Pipeline 可以与各种第三方库集成，如用于数据分析的 Pandas，用于机器学习的 Scikit-learn 等。

8. 总结

Item Pipelines 是 Scrapy 框架中一个非常重要的组成部分，它提供了一种灵活且强大的机制来处理爬取的数据。通过合理地使用 Item Pipelines，可以有效地清洗、验证、去重和存储数据，从而提高爬虫的质量和效率。理解和掌握 Item Pipelines 的使用是成为一名优秀的 Scrapy 开发者所必需的。