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

我有一个包含超过 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 推送到 DB(连同其他东西)-> 重复

100,000 个项目需要 30 分钟。瓶颈是 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 次。这会有什么不同吗？

请您参考如下方法:

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

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")