在Python中高效检查数百万个图像URL

问题描述

我有一个超过300万个项目行的tsv文件。 每个项目都有一个id，组和一个url，并且组列已排序。

即

x1    gr1    {some url}/x1.jpg
x2    gr1    {some url}/x2.jpg
x3    gr2    {some url}/x1.jpg  

我将其加载到python脚本中，并且需要在将组中所有项目的URL加载到数据库中之前检查状态200 OK。 我想到了使用进程并对每个进程进行URL检查，（我对此没有太多经验，因此不确定它是否是个好主意）

我的逻辑atm：用gr1的项目填充数组a1->将a1中的每个项目传递给新进程->该进程检查200->如果可以，则将其放入数组a2中->检查a1中的所有项目后，将a2推入DB（以及其他东西）->重复

大约需要30分钟才能处理100,000个项目。 瓶颈是URL检查。 如果不检查URL，则与之相比，该脚本的运行速度很快。 至今：

import csv
import re
import requests
import multiprocessing
from pymongo import MongoClient
import sys

#Load in Data
f = open('../tsvsorttest.tsv', 'rb')
reader = csv.reader(f, delimiter='\n')

#Get the first group name
currGroup = re.split(r'\t', next(reader)[0].decode('utf8'))[1]
currGroupNum = 0 
items = []
checkedItems = []

#Method that checks the URL, if its 200, add to newItems
def check_url(newitem):
    if requests.get(newitem['image_url']).status_code is 200:
        print('got an ok!')
        checkedItems.append(newitem)
    global num_left
    num_left -= 1


def clear_img(checkitems):
    for thisItem in checkitems:
        p = multiprocessing.Process(target=check_url(thisItem))
        p.start()

#Start the loop, use i to keep track of the iteration count
for i, utf8_row in enumerate(reader):
    unicode_row = utf8_row[0].decode('utf8')

    x = re.split(r'\t', unicode_row)

    item = {"id": x[0],
            "group": x[1],
            "item_url": x[2]
            }
    if currGroup != x[1]:
        y = len(items)
        print('items length is ' + str(y))

        #Dont want single item groups
        if y > 1:
            print 'beginning url checks'
            num_left = len(items)


            clear_img(items)
            while num_left is not 0:
                print 'Waiting'

            num_left = 0
            batch = {"vran": currGroup,
                     "bnum": currGroupNum,
                     "items": newItems,
                     }
            if len(checkedItems) > 0:
                batches.insert_one(batch)
                currGroupNum += 1

        currGroup = x[1]
        items = []
        checkedItems = []

    items.append(item)

    if i % 100 == 0:
        print "Milestone: " + str(i)

print "done"

其他注意事项：将原始Tsv拆分成类似30个单独的tsv文件，并并行运行批处理脚本30次。 这会有所作为吗？

已经提到您应该尝试使用HEAD而不是GET 。 这样可以避免下载图像。 而且，您似乎在为每个请求生成单独的进程 ，这也效率很低。

我不认为在性能方面确实不需要使用asyncio。 使用普通线程池（甚至不是进程池）的解决方案有点麻烦，恕我直言：)另外，它在Python 2.7中可用。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import csv
from collections import defaultdict

def read_rows(file):
    with open(file) as f_in:
        return [row for row in csv.reader(f_in, delimiter='\t')]

def check_url(inp):
    """Gets called by workers in thread pool. Checks for existence of URL."""
    id, grp, url = inp
    def chk():
        try:
            return requests.head(url).status_code == 200
        except IOError as e:
            return False
    return (id, grp, url, chk())

if __name__ == '__main__':
    d = defaultdict(lambda: [])
    with ThreadPoolExecutor(max_workers=20) as executor:
        future_to_input = {executor.submit(check_url, inp): inp for inp in read_rows('urls.txt')}
        for future in as_completed(future_to_input):
            id, grp, url, res = future.result()
            d[grp].append((id, url, res))
    # do something with your d (e.g. sort appropriately, filter those with len(d[grp]) <= 1, ...)
    for g, bs in d.items():
        print(g)
        for id, url, res in bs:
            print("  %s %5s %s" % (id, res, url))

如您所见，我分别处理CSV输入的每一行，然后对结果进行分组（使用d ），而不是对输入进行分组。 我想主要是口味问题。 您可能想玩max_workers=20并可能增加它。