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 绑定(bind)操作(如解析/搜索 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])

A lazier version of map().

The chunksize argument is the same as the one used by the map() method. For very long iterables using a large value for chunksize can make the job complete much faster than using the default value of 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 行，这应该更易于管理。