将网络抓取写入文件分发到 Python 中的并行进程？答案

【问题标题】：Distribute web-scraping write-to-file to parallel processes in Python?将网络抓取写入文件分发到 Python 中的并行进程？
【发布时间】：2016-02-08 14:23:28
【问题描述】：

我正在从一个网站上抓取一些 JSON 数据，并且需要这样做大约 50,000 次（所有数据都是针对 3 年内不同的邮政编码）。我为大约 1,000 次调用超时了程序，每次调用的平均时间为 0.25 秒，整个范围（全部 50,000 次调用）的运行时间约为 3.5 小时。

如何将这个进程分布到我的所有内核上？我的代码的核心几乎是这样的：

with open("U:/dailyweather.txt", "r+") as f:
    f.write("var1\tvar2\tvar3\tvar4\tvar5\tvar6\tvar7\tvar8\tvar9\n")
    writeData(zips, zip_weather_links, daypart)

writeData() 看起来像这样：

def writeData(zipcodes, links, dayparttime):
    for z in zipcodes:
        for pair in links:
            ## do some logic ##
            f.write("%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n" % (var1, var2, var3, var4, var5, 
                                                              var6, var7, var8, var9))

zips 看起来像这样：

zips = ['55111', '56789', '68111', ...]

而zip_weather_links 只是每个邮政编码的（URL，日期）字典：

zip_weather_links['55111'] = [('https://website.com/55111/data', datetime.datetime(2013, 1, 1, 0, 0, 0), ...]

如何使用Pool 或multiprocessing 分发此内容？或者分发甚至会节省时间？

【问题讨论】：

从 web 服务获取 JSON 数据的方法在哪里？那是不写入文件的瓶颈。您需要为该方法实现 Pool。
该方法包含在 writeData 中，在另一个方法 matchTimeandWeather() 中。基本上是这样的：dat = json.loads(page)。我真的只想将 1/4 的负载分配给每个核心（即让一个核心负责邮政编码 a、b 和 c，而另一个核心负责 d、e 和 f，等等。

标签： python parallel-processing web-scraping multiprocessing pool

【解决方案1】：

您希望“将网络抓取写入文件分发到 Python 中的并行进程”。首先，让我们看看 Web-Scraping 使用时间最多的地方。

HTTP 请求的延迟远高于硬盘。 Link: Latency comparison。对硬盘的小写入比大写入慢得多。 SSD 具有更高的随机写入速度，因此这种影响不会太大。

分发 HTTP 请求
收集所有结果
一次将所有结果写入磁盘

一些带有IPython parallel的示例代码：

from ipyparallel import Client
import requests
rc = Client()
lview = rc.load_balanced_view()
worklist = ['http://xkcd.com/614/info.0.json',
            'http://xkcd.com/613/info.0.json']

@lview.parallel()
def get_webdata(w):
    import requests
    r = requests.get(w)
    if not r.status_code == 200:
         return (w, r.status_code,)
    return (w, r.json(),)

#get_webdata will be called once with every element of the worklist
proc = get_webdata.map(worklist) 
results = proc.get()
# results is a list with all the return values
print(results[1])
# TODO: write the results to disk

你必须先启动 IPython 并行工作者：

(py35)River:~ rene$ ipcluster start -n 20

【讨论】：