爬虫工作量由小到大的思维转变---<第十三章 Scrapy之pipelines分离的思考>

2023-12-18 14:50:40

前言:

收到留言: "我的爬取的数据处理有点大,scrapy抓网页挺快,处理数据慢了!"

-----针对这位粉丝留言,我只想说:'你那培训班老师可能给你漏了课程! 大概你们上课讲的案例属于demo,他教了你一些基本操作,但他没有对相关业务对你讲透! 你研究一下pipelines,或者看我现在给你讲的.

正文

首先,你要清楚,当在Scrapy框架中,pipelines是顺序执行的,对item的处理通常是同步进行。

这时候,你要分析2件事:

1.我的数据要不要清洗

2.我的数据准备怎么存储

分开讲:

1.我的数据要不要清洗:

如果需要清洗,item的数据里比较多,我建议你转一下pd.dataframe;这样,会比正常运算要快得多;然后,给你3条建议:

  1. 避免在循环内使用 df.apply():--->? ? apply() 是行或列级别的操作函数,效率相对较低。如果可以,尝试用更高效的Pandas内建函数代替,比如使用逻辑运算与 numpy 的向量化操作。
  2. 对于字符串处理,如果数据量很大,应当尽量使用向量化方法,例如 .str 方法或其他Pandas字符串操作代替 lambda 函数。
  3. 当创建新的列时,用条件表达式替代 .apply(lambda) 可以获得更好的性能,条件表达式在Pandas中是向量化的。
如果pandas处理之后,不满足:
分离繁重操作:
如果有些操作很繁重,可以将它们移动到Scrapy的middleware或者扩展来进行,这样可能有助于提高item pipeline的处理速度。这时候,你就可以通过外部自己写一个多线程/多进程来处理你的数据工作!
当然,处理item的数据清理工作,我建议你用:
ItemAdapter
什么是ItemAdapter?

-它是一个包装类,允许我们以一致的方式处理不同种类的数据结构,例如dict、scrapy.Item以及自定义的数据类。无论内部的数据存储格式如何,ItemAdapter都能让我们同等的获取和设置Item中的字段值。


ItemAdapter的使用场景

ItemAdapter特别适用于编写更通用的Pipeline代码。无论传入的Item是Scrapy的Item实例还是普通的dict,甚至是自定义的类实例,你都可以使用相同的方法来处理它们。这样的设计大大提升了代码的复用性和可维护性。

案例:
import scrapy
from itemadapter import ItemAdapter
import pandas as pd
import numpy

class JihaiPipeline:
    def open_spider(self, spider):
        # 初始化工作,例如连接数据库
        pass

    def close_spider(self, spider):
        # 清理工作,例如关闭数据库连接
        pass

    def process_item(self, item, spider):
        # 使用ItemAdapter包装item
        adapter = ItemAdapter(item)
        
        # 进行数据处理...
        # 例如,假设我们需要给所有Item添加一个新字段
        adapter['new_field'] = '丢一个新的字段进去'
        
        # 处理完后,返回item
        return item

在上面的代码中,我们没有直接操作原始的item对象,而是将其通过ItemAdapter(item)包装起来。然后就可以像操作字典一样,通过adapter['new_field']来设置新字段。在管道中修改完数据后,可以直接将Item传递到下一个管道。

ItemAdapter中的向量化操作

对于爬虫项目,可能需要对数据进行更复杂的清洗和转换操作。在Pandas的帮助下,我们可以执行向量化的数据处理工作,这是一种高效处理数据的方式。通过Pandas,利用DataFrame进行复杂的数据清洗和分析变得相当简便

案例:
class JihaiPipeline:
    # ...之前的方法...

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)

        # 假设我们的item有一个成绩的列表需要处理
        grades = adapter.get('grades', [])

        # 使用Pandas创建DataFrame
        df = pd.DataFrame(grades)
        # 执行一些复杂的计算操作,例如计算平均分
        adapter['average_grade'] = df['score'].mean()

        # 返回处理后的item
        return item

在这个例子中,我们先获取了成绩列表,然后使用这个列表创建了一个Pandas DataFrame。之后我们就可以利用DataFrame提供的方法进行各种操作,比如这里计算了一个平均分成绩,然后将其添加到了item中。

小总结:

ItemAdapter提供了一个透明的方式来处理项,帮助你更简单地编写与项结构无关的代码。与Pandas结合使用,它也使得在Scrapy中进行复杂数据处理成为可能。记住,一致性、可读性和可维护性是编写高质量爬虫代码时的关键点。


2.我的数据准备怎么存储?

?

如果你的数据比较单一,你直接存(就跟你老师教你的那样!) 如果你的数据已经到达了你的瓶颈,你最好做个分离;然后看我之前的文章,例如:存入sql--->你首先要想到的就是异步!

在Scrapy中,最佳实践通常是将数据处理(清洗、转换等)与数据存储(写入数据库等)分离。这为你的数据处理流水线提供了更好的组织结构和可扩展性。每个Pipeline应该只负责一个操作或一组相关操作。这样做的好处是:

1. 职责分离:这使得每个pipeline的职责更清晰。如果以后需要更改存储逻辑,只需要更改保存到SQL的pipeline,而不需要触及数据处理的pipeline。
2. 模块化:如果在将来需要将数据存储到不同的后端(例如不同的数据库,或者文件系统等),你可以简单地添加一个新的pipeline来处理这种情况,而不是更改现有代码。
3. 可维护性:代码维护更简单,因为数据清洗和存储是分开的,错误更容易追踪,代码更容易调试。
4. 可测试性:独立的pipeline更容易进行单元测试。

既然已经完成了数据处理,并且将结果整理成了待存储的格式,接下来的逻辑步骤是将这些数据保存到SQL数据库。创建一个新的Pipeline类专门用于与SQL数据库的交互,这样,你的 `XXXPipeline` 负责处理数据,并将处理后的数据传递给稍后在settings.py文件中定义优先级更低的SQL存储pipeline。

下面是创建一个专门用于存储数据到SQL数据库的pipeline的简单例子(要异步,往前看我文章有介绍):

# sql_pipeline.py

import scrapy
from scrapy import Item
from itemadapter import ItemAdapter

class SQLStorePipeline:
    def open_spider(self, spider):
        # 这里设置数据库连接
        self.connection = create_connection_to_database()

    def close_spider(self, spider):
        # 关闭数据库连接
        self.connection.close()

    def process_item(self, item, spider):
        # 提取ItemAdapter
        adapter = ItemAdapter(item)
        # 保存到数据库的逻辑
        save_to_database(self.connection, adapter.as_dict())
        return item  # 注意,返回item是为了允许多个pipeline

def create_connection_to_database():
    # 创建数据库链接逻辑
    pass

def save_to_database(connection, item_data):
    # 将item数据保存到数据库的逻辑
    pass

在`settings.py`文件中,您需要确保新的`SQLStorePipeline`在`XXXPipeline`之后执行。这可以通过为它们分配不同的`ITEM_PIPELINES`值来实现:

# settings.py

ITEM_PIPELINES = {
    'myproject.pipelines.XXXPipeline': 300,  #处理数据清理的
    'myproject.pipelines.SQLStorePipeline': 800,   #存储的
}

这样,每个item首先通过`JihaiPipeline`进行处理,然后再通过`SQLStorePipeline`进行存储。

通过这种方式,您既保持了pipeline的职责分割,又为后续的维护和可能的扩展性打下了良好的基础。如果有多个数据存储或处理需求,遵循这种模式是非常有好处的。

总结:

你就记住,如果你的item数据量比较大,一定要分离! 分完了,很多都能处理了! 另外,你记得itemAdapter的用法~ 他应该算是一个引子,透过他~你写着写着就会冒出很多怪招出来~ 然后,再不行,你就进行分布式! 反正你的业务已经模块化了,拿一个机器专门清理,拿一个机器专门存储~或者,丢到中间件,甩到外部去做多线程处理!这样,在爬虫过程中,对数据的清理和存储的工作量,就能被划分掉,不就轻了么...? ?

请你看到这文章,给我点个赞!!

(让我知道你来了)

文章来源:https://blog.csdn.net/m0_56758840/article/details/135052658
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。