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

Question

我有一个包含超过 300 万行项目的 tsv 文件。每个 Item 都有一个 id、group 和一个 url，并且对 group 列进行了排序。

即

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 推送到数据库（连同其他东西）-> 重复

这需要 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 次。这会有所不同吗？

Answer 1

1. 由于您不需要实际图像，因此使用 HEAD 请求应该会提高速度。如果响应既不是 200 也不是 404 可能不允许 HEAD (405)，您只需使用 GET 请求重试即可。
2. 您目前正在等待当前组完成，然后再开始任何新任务。通常，最好始终保持相同数量的运行请求大致相同。此外，您可能希望大幅增加工作人员池 - 由于任务主要受 I/O 限制，但我建议您按照 3 的方式进行操作（即异步 I/O）。
3. 如果您愿意使用 Python 3，您可以通过使用 https://pypi.python.org/pypi/aiohttp 来利用对异步 I/O (https://docs.python.org/3/library/asyncio.html) 的出色支持：

import asyncio
from aiohttp import ClientSession, Timeout
import csv
import re
from threading import Thread
from queue import Queue
from time import sleep

async def check(url, session):
    try:
        with Timeout(10):
            async with session.head(url) as response:
                if response.status == 200:
                    return True
                elif response.status == 404:
                    return False
                else:
                    async with session.get(url) as response:
                        return (response.status == 200)
    except:
        return False



def worker(q):
    while True:
        f = q.get()
        try:
            f()
        except Exception as e:
            print(e)
        q.task_done()

q = Queue()
for i in range(4):
     t = Thread(target=worker,args=(q,))
     t.daemon = True
     t.start()

def item_ok(url):
    #Do something
    sleep(0.5)
    pass

def item_failed(url):
    #Do something
    sleep(0.5)
    pass

def group_done(name,g):
    print("group %s with %d items done (%d failed)\n" %
          (name,g['total'],g['fail']))

async def bound_check(sem, item, session, groups):
    async with sem:
        g = groups[item["group"]]
        if (await check(item["item_url"], session)):
            g["success"] += 1
            q.put(lambda: item_ok(item["item_url"]))
        else:
            g["fail"] += 1
            q.put(lambda: item_failed(item["item_url"]))
        if g["success"] + g["fail"] == g['total']:
            q.put(lambda: group_done(item['group'],g))
        bound_check.processed += 1
        if bound_check.processed % 100 == 0:
            print ("Milestone: %d\n" % bound_check.processed)

bound_check.processed = 0

groups = {}

async def run(max_pending=1000):
    #Choose such that you do not run out of FDs
    sem = asyncio.Semaphore(max_pending)

    f = open('./test.tsv', 'r',encoding = 'utf8')
    reader = csv.reader(f, delimiter='\n')

    tasks = []

    async with ClientSession() as session:
        for _, utf8_row in enumerate(reader):
            unicode_row = utf8_row[0]
            x = re.split(r'\t', unicode_row)
            item = {"id": x[0],"group": x[1],"item_url": x[2]}
            if not item["group"] in groups:
                groups[item["group"]] = {'total'    : 1,
                                         'success'  : 0,
                                         'fail'     : 0,
                                         'items'    : [item]}
            else:
                groups[item["group"]]['total'] += 1
                groups[item["group"]]['items'].append(item)
            task = asyncio.ensure_future(bound_check(sem, item, session, groups))
            tasks.append(task)

        responses = asyncio.gather(*tasks)
        await responses

loop = asyncio.get_event_loop()
loop.run_until_complete(run())
q.join()

print("Done")

Answer 2

已经提到你应该尝试使用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 并可能增加它。