Python/BeautifulSoup 中的多线程抓取根本没有加速

Question

我有一个 csv 文件（“SomeSiteValidURLs.csv”），其中列出了我需要抓取的所有链接。该代码正在运行，并将通过 csv 中的 url，抓取信息并记录/保存在另一个 csv 文件（“Output.csv”）中。但是，由于我计划为网站的大部分内容（>10,000,000 个页面）执行此操作，因此速度很重要。对于每个链接，爬取信息并将信息保存到 csv 大约需要 1 秒，这对于项目的规模来说太慢了。所以我加入了多线程模块，令我惊讶的是它根本没有加速，它仍然需要 1 个人链接。我做错什么了吗？有没有其他方法可以加快处理速度？

没有多线程：

import urllib2
import csv
from bs4 import BeautifulSoup
import threading

def crawlToCSV(FileName):

    with open(FileName, "rb") as f:
        for URLrecords in f:

            OpenSomeSiteURL = urllib2.urlopen(URLrecords)
            Soup_SomeSite = BeautifulSoup(OpenSomeSiteURL, "lxml")
            OpenSomeSiteURL.close()

            tbodyTags = Soup_SomeSite.find("tbody")
            trTags = tbodyTags.find_all("tr", class_="result-item ")

            placeHolder = []

            for trTag in trTags:
                tdTags = trTag.find("td", class_="result-value")
                tdTags_string = tdTags.string
                placeHolder.append(tdTags_string)

            with open("Output.csv", "ab") as f:
                writeFile = csv.writer(f)
                writeFile.writerow(placeHolder)

crawltoCSV("SomeSiteValidURLs.csv")

多线程：

import urllib2
import csv
from bs4 import BeautifulSoup
import threading

def crawlToCSV(FileName):

    with open(FileName, "rb") as f:
        for URLrecords in f:

            OpenSomeSiteURL = urllib2.urlopen(URLrecords)
            Soup_SomeSite = BeautifulSoup(OpenSomeSiteURL, "lxml")
            OpenSomeSiteURL.close()

            tbodyTags = Soup_SomeSite.find("tbody")
            trTags = tbodyTags.find_all("tr", class_="result-item ")

            placeHolder = []

            for trTag in trTags:
                tdTags = trTag.find("td", class_="result-value")
                tdTags_string = tdTags.string
                placeHolder.append(tdTags_string)

            with open("Output.csv", "ab") as f:
                writeFile = csv.writer(f)
                writeFile.writerow(placeHolder)

fileName = "SomeSiteValidURLs.csv"

if __name__ == "__main__":
    t = threading.Thread(target=crawlToCSV, args=(fileName, ))
    t.start()
    t.join()

Answer 1

您没有正确并行化它。您真正想要做的是让您的 for 循环中正在完成的工作同时发生在许多工作人员身上。现在，您正在将所有工作转移到一个后台线程中，该线程同步完成整个工作。这根本不会提高性能（实际上只会稍微伤害它）。

这是一个使用 ThreadPool 并行化网络操作和解析的示例。尝试一次跨多个线程写入 csv 文件是不安全的，因此我们将本应写回父级的数据返回，并让父级在最后将所有结果写入文件。

import urllib2
import csv
from bs4 import BeautifulSoup
from multiprocessing.dummy import Pool  # This is a thread-based Pool
from multiprocessing import cpu_count

def crawlToCSV(URLrecord):
    OpenSomeSiteURL = urllib2.urlopen(URLrecord)
    Soup_SomeSite = BeautifulSoup(OpenSomeSiteURL, "lxml")
    OpenSomeSiteURL.close()

    tbodyTags = Soup_SomeSite.find("tbody")
    trTags = tbodyTags.find_all("tr", class_="result-item ")

    placeHolder = []

    for trTag in trTags:
        tdTags = trTag.find("td", class_="result-value")
        tdTags_string = tdTags.string
        placeHolder.append(tdTags_string)

    return placeHolder


if __name__ == "__main__":
    fileName = "SomeSiteValidURLs.csv"
    pool = Pool(cpu_count() * 2)  # Creates a Pool with cpu_count * 2 threads.
    with open(fileName, "rb") as f:
        results = pool.map(crawlToCSV, f)  # results is a list of all the placeHolder lists returned from each call to crawlToCSV
    with open("Output.csv", "ab") as f:
        writeFile = csv.writer(f)
        for result in results:
            writeFile.writerow(result)

请注意，在 Python 中，线程实际上只加速 I/O 操作 - 由于 GIL，CPU 绑定操作（如解析/搜索BeautifulSoup 正在执行）实际上不能通过线程并行完成，因为一次只有一个线程可以执行基于 CPU 的操作。因此，您可能仍然看不到这种方法所希望的速度。当您需要在 Python 中加速 CPU 密集型操作时，您需要使用多个进程而不是线程。幸运的是，您可以很容易地看到这个脚本是如何使用多个进程而不是多个线程执行的；只需将from multiprocessing.dummy import Pool 更改为from multiprocessing import Pool。无需进行其他更改。

编辑：

如果您需要将其扩展到具有 10,000,000 行的文件，则需要稍微调整此代码 - Pool.map 将您传递给它的可迭代对象转换为列表，然后再将其发送到您的工人，这显然不适用于 10,000,000 个条目列表；将整个内容保存在内存中可能会使您的系统陷入困境。将所有结果存储在列表中的问题相同。相反，您应该使用Pool.imap:

imap(func, iterable[, chunksize])

map() 的惰性版本。

chunksize 参数与 map() 使用的参数相同 方法。对于很长的迭代，使用较大的 chunksize 值可以 使作业完成比使用默认值 1 快得多。

if __name__ == "__main__":
    fileName = "SomeSiteValidURLs.csv"
    FILE_LINES = 10000000
    NUM_WORKERS = cpu_count() * 2
    chunksize = FILE_LINES // NUM_WORKERS * 4   # Try to get a good chunksize. You're probably going to have to tweak this, though. Try smaller and lower values and see how performance changes.
    pool = Pool(NUM_WORKERS)

    with open(fileName, "rb") as f:
        result_iter = pool.imap(crawlToCSV, f)
    with open("Output.csv", "ab") as f:
        writeFile = csv.writer(f)
        for result in result_iter:  # lazily iterate over results.
            writeFile.writerow(result)

使用imap，我们不会一次将所有f 放入内存，也不会一次将所有结果存储在内存中。我们记忆中最多的是chunksize 行f，应该更易于管理。