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

前言:

收到留言: "我的爬取的数据处理有点大,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 numpyclass JihaiPipeline:def open_spider(self, spider):# 初始化工作，例如连接数据库passdef close_spider(self, spider):# 清理工作，例如关闭数据库连接passdef process_item(self, item, spider):# 使用ItemAdapter包装itemadapter = ItemAdapter(item)# 进行数据处理...# 例如，假设我们需要给所有Item添加一个新字段adapter['new_field'] = '丢一个新的字段进去'# 处理完后，返回itemreturn 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创建DataFramedf = pd.DataFrame(grades)# 执行一些复杂的计算操作，例如计算平均分adapter['average_grade'] = df['score'].mean()# 返回处理后的itemreturn 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.pyimport scrapy
from scrapy import Item
from itemadapter import ItemAdapterclass 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):# 提取ItemAdapteradapter = ItemAdapter(item)# 保存到数据库的逻辑save_to_database(self.connection, adapter.as_dict())return item  # 注意，返回item是为了允许多个pipelinedef create_connection_to_database():# 创建数据库链接逻辑passdef save_to_database(connection, item_data):# 将item数据保存到数据库的逻辑pass

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

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

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

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

总结:

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

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

(让我知道你来了)