4.3 数据清洗与预处理

4.3 数据清洗与预处理

Scrapy实践应用：数据清洗与预处理详解

1. 数据清洗与预处理的重要性

数据清洗与预处理的目的是提高数据的质量和可用性，主要体现在以下几个方面：

• 提高数据准确性： 移除错误或不完整的数据，纠正拼写错误和格式错误。

• 提高数据一致性： 统一数据格式、单位和编码，避免数据冲突和歧义。

• 提高数据完整性： 填充缺失值或删除不完整记录，确保数据的完整性。

• 提高数据可用性： 将数据转换为结构化格式，方便后续分析和存储。

2. Scrapy中的数据清洗与预处理流程

在Scrapy中，数据清洗与预处理通常在Item Pipeline中进行。Item Pipeline是一个组件，负责处理Spider提取的Item。它接收Item，执行一系列处理步骤，然后将处理后的Item传递给下一个Pipeline或存储到数据库。

以下是一个简化的数据清洗与预处理流程图：

3. 数据清洗与预处理的常用技术

以下是一些常用的数据清洗与预处理技术，以及在Scrapy中的应用：

3.1 HTML标签去除

从网页抓取的数据通常包含HTML标签，这些标签对于数据分析来说是无用的噪音。可以使用以下方法去除HTML标签：

• 使用lxml或BeautifulSoup解析HTML： 这两个库可以将HTML文档解析成树形结构，方便提取文本内容。

• 使用正则表达式： 正则表达式可以匹配和替换HTML标签。

• 使用Scrapy内置的remove_tags处理器： 这是一个方便的内置函数，可以快速去除HTML标签。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst, Join, Identity
from w3lib.html import remove_tags
def remove_html_tags(value):
    return remove_tags(value)
class ProductItem(Item):
    name = Field(output_processor=TakeFirst())
    description = Field(input_processor=MapCompose(remove_html_tags, str.strip), output_processor=Join())
    price = Field(output_processor=TakeFirst())
class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            loader = ProductLoader(item=ProductItem(), selector=product)
            loader.add_css("name", ".product-name::text")
            loader.add_css("description", ".product-description") # 假设描述包含HTML标签
            loader.add_css("price", ".product-price::text")
            yield loader.load_item()
​

代码详解：

• remove_html_tags 函数使用 w3lib.html.remove_tags 去除HTML标签。

• ProductItem 定义了要抓取的数据字段。

• description 字段的 input_processor 使用 MapCompose 将 remove_html_tags 应用于每个描述文本，并使用 str.strip 去除首尾空格。output_processor 使用 Join 将处理后的文本连接成一个字符串。

• ProductLoader 用于填充 ProductItem，default_output_processor 设置为 TakeFirst，表示只取第一个值。

• MySpider 是一个示例Spider，从 http://example.com/products 抓取数据，并使用 ProductLoader 加载数据。

3.2 空格和特殊字符去除

网页文本中可能包含多余的空格、换行符、制表符等特殊字符。可以使用以下方法去除这些字符：

• 使用str.strip()方法： 去除字符串首尾的空格。

• 使用str.replace()方法： 替换特定的字符。

• 使用正则表达式： 匹配和替换特定的字符。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst
def remove_whitespace(value):
    return value.strip()
def replace_newline(value):
    return value.replace('\n', '')
class ProductItem(Item):
    name = Field(input_processor=MapCompose(remove_whitespace), output_processor=TakeFirst())
    address = Field(input_processor=MapCompose(remove_whitespace, replace_newline), output_processor=TakeFirst())
class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            loader = ProductLoader(item=ProductItem(), selector=product)
            loader.add_css("name", ".product-name::text")
            loader.add_css("address", ".product-address::text") # 假设地址包含换行符和空格
            yield loader.load_item()
​

代码详解：

• remove_whitespace 函数使用 str.strip() 去除字符串首尾的空格。

• replace_newline 函数使用 str.replace('\n', '') 替换换行符为空字符串。

• name 字段的 input_processor 使用 MapCompose 将 remove_whitespace 应用于每个名称文本。

• address 字段的 input_processor 使用 MapCompose 将 remove_whitespace 和 replace_newline 应用于每个地址文本。

3.3 数据类型转换

从网页抓取的数据通常是字符串类型，需要根据实际需求转换为其他数据类型，例如整数、浮点数、日期等。可以使用以下方法进行数据类型转换：

• 使用int()、float()、datetime.datetime.strptime()等函数： 将字符串转换为相应的数据类型。

• 使用try-except块处理转换错误： 避免因数据格式错误导致程序崩溃。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst
import datetime
def convert_to_int(value):
    try:
        return int(value)
    except ValueError:
        return None  # 或者返回一个默认值
def convert_to_datetime(value, formats=('%Y-%m-%d', '%Y/%m/%d')):
    for fmt in formats:
        try:
            return datetime.datetime.strptime(value, fmt)
        except ValueError:
            pass
    return None # 或者返回一个默认值
class ProductItem(Item):
    product_id = Field(input_processor=MapCompose(convert_to_int), output_processor=TakeFirst())
    release_date = Field(input_processor=MapCompose(convert_to_datetime), output_processor=TakeFirst())
    price = Field(input_processor=MapCompose(lambda x: float(x.replace('$', '').replace(',', '') if x else None)), output_processor=TakeFirst())
class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            loader = ProductLoader(item=ProductItem(), selector=product)
            loader.add_css("product_id", ".product-id::text")
            loader.add_css("release_date", ".product-release-date::text")
            loader.add_css("price", ".product-price::text")
            yield loader.load_item()
​

代码详解：

• convert_to_int 函数尝试将字符串转换为整数，如果转换失败则返回 None。

• convert_to_datetime 函数尝试将字符串转换为日期时间，支持多种日期格式，如果转换失败则返回 None。

• price 字段的 input_processor 使用 lambda 函数去除 $ 和 , 符号，然后将字符串转换为浮点数，如果为空则返回 None。

• 使用 try-except 块可以优雅地处理数据类型转换错误，避免程序崩溃。

3.4 数据规范化

数据规范化是将数据缩放到一个特定的范围，例如[0, 1]或[-1, 1]。这可以消除数据量纲的影响，提高模型的性能。常用的数据规范化方法包括：

• Min-Max Scaling： 将数据缩放到[0, 1]范围。

• Z-Score Standardization： 将数据缩放到均值为0，标准差为1的分布。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst
def min_max_scaling(value, min_val, max_val):
    try:
        value = float(value)
        return (value - min_val) / (max_val - min_val)
    except (ValueError, TypeError):
        return None
class ProductItem(Item):
    rating = Field(input_processor=MapCompose(lambda x: min_max_scaling(x, 1, 5)), output_processor=TakeFirst())
class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            loader = ProductLoader(item=ProductItem(), selector=product)
            loader.add_css("rating", ".product-rating::text")
            yield loader.load_item()
​

代码详解：

• min_max_scaling 函数将数据缩放到[0, 1]范围，需要指定数据的最小值和最大值。

• rating 字段的 input_processor 使用 lambda 函数将 min_max_scaling 应用于每个评分值，假设评分范围是[1, 5]。

3.5 数据去重

在爬取过程中，可能会抓取到重复的数据。可以使用以下方法去除重复数据：

• 在Item Pipeline中使用集合（Set）或数据库： 将已经处理过的Item的唯一标识符（例如URL）存储在集合或数据库中，当遇到重复的Item时，直接丢弃。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.exceptions import DropItem
class ProductItem(Item):
    name = Field()
    url = Field()
class DuplicatesPipeline:
    def __init__(self):
        self.urls_seen = set()
    def process_item(self, item, spider):
        if item['url'] in self.urls_seen:
            raise DropItem("Duplicate item found: %s" % item)
        else:
            self.urls_seen.add(item['url'])
            return item
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            item = ProductItem()
            item['name'] = product.css(".product-name::text").get()
            item['url'] = response.urljoin(product.css("a::attr(href)").get())
            yield item
​

代码详解：

• DuplicatesPipeline 维护一个 urls_seen 集合，用于存储已经处理过的URL。

• process_item 方法检查当前Item的URL是否已经存在于 urls_seen 集合中，如果存在则抛出 DropItem 异常，丢弃该Item；否则将URL添加到 urls_seen 集合中，并返回该Item。

• 需要在 settings.py 中启用 DuplicatesPipeline，例如：ITEM_PIPELINES = {'myproject.pipelines.DuplicatesPipeline': 300}。

3.6 缺失值处理

在某些情况下，抓取的数据可能包含缺失值。可以使用以下方法处理缺失值：

• 填充缺失值： 使用平均值、中位数、众数等统计量填充缺失值。

• 删除包含缺失值的记录： 如果缺失值过多，可以考虑删除包含缺失值的记录。

代码示例：

import scrapy
from scrapy.item import Item, Field
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst
def fill_missing_price(value, default_price=0.0):
    try:
        return float(value)
    except (ValueError, TypeError):
        return default_price
class ProductItem(Item):
    price = Field(input_processor=MapCompose(fill_missing_price), output_processor=TakeFirst())
class ProductLoader(ItemLoader):
    default_output_processor = TakeFirst()
class MySpider(scrapy.Spider):
    name = "myspider"
    start_urls = ["http://example.com/products"]
    def parse(self, response):
        for product in response.css(".product"):
            loader = ProductLoader(item=ProductItem(), selector=product)
            loader.add_css("price", ".product-price::text")
            yield loader.load_item()
​

代码详解：

• fill_missing_price 函数尝试将字符串转换为浮点数，如果转换失败则返回默认价格 0.0。

• price 字段的 input_processor 使用 MapCompose 将 fill_missing_price 应用于每个价格值。

4. Item Loader的使用

Scrapy的Item Loader提供了一种方便的方式来填充Item，并进行数据清洗和预处理。Item Loader可以自动应用预定义的处理器，将抓取的数据转换为所需的格式。

以下是Item Loader的常用组件：

• Input Processors： 在数据被添加到Item之前应用的处理函数。

• Output Processors： 在数据被从Item中提取出来之后应用的处理函数。

• MapCompose： 将多个处理器函数依次应用于每个输入值。

• TakeFirst： 从列表中提取第一个非空值。

• Join： 将列表中的所有值连接成一个字符串。

• Identity: 不做任何处理，直接返回原始值。

5. 总结

数据清洗与预处理是Scrapy爬虫项目中不可或缺的环节。通过使用Item Pipeline和Item Loader，可以方便地进行各种数据清洗和预处理操作，将抓取的数据转化为结构化、可用的数据。本文介绍了常用的数据清洗与预处理技术，并提供了相应的代码示例。在实际项目中，需要根据具体的需求选择合适的技术和方法。