Google appengine: производительность очереди задач

Question

В настоящее время у меня есть приложение, работающее на appengine, и я выполняю несколько заданий с использованием отложенной библиотеки, некоторые из этих задач запускаются ежедневно, а некоторые из них выполняются раз в месяц. Большинство этих задач запрашивают Datastore для получения документов, а затем сохраняют объекты в индексе (Search API). Некоторые из этих таблиц заменяются ежемесячно, и я должен выполнять эти задачи для всех объектов (4 ~ 5M).

Один Exemple такой задачи является:

def addCompaniesToIndex(cursor=None, n_entities=0, mindate=None): 
    #get index 
    BATCH_SIZE = 200 
    cps, next_cursor, more = Company.query().\ 
            fetch_page(BATCH_SIZE, 
               start_cursor=cursor) 

    doc_list = [] 

    for i in range(0, len(cps)): 
     cp = cps[i] 
     #create a Index Document using the Datastore entity 
     #this document has only about 5 text fields and one date field 
     cp_doc = getCompanyDocument(cp) 
     doc_list.append(cp_doc) 

    index = search.Index(name='Company') 
    index.put(doc_list) 

    n_entities += len(doc_list) 

    if more: 
     logging.debug('Company: %d added to index', n_entities) 
     #to_put[:] = [] 
     doc_list[:] = [] 
     deferred.defer(addCompaniesToIndex, 
         cursor=next_cursor, 
         n_entities=n_entities, 
         mindate=mindate) 
    else: 
     logging.debug('Finished Company index creation (%d processed)', n_entities) 

Когда я запускаю только одну задачи, выполнение занимает около 4-5s за отсроченную задачу, поэтому индексировать мои объекты 5M займут около 35 часов.

Другое дело, что когда я запускаю обновление по другому индексу (например, одно из ежедневных обновлений) с использованием другой отложенной задачи в одной очереди, обе выполняются намного медленнее. И начните принимать около 10-15 секунд за отложенный вызов, который просто невыносим.

Мой вопрос: есть ли способ сделать это быстрее и масштабировать очередь push для более чем одного задания, выполняемого каждый раз? Или я должен использовать другой подход для этой проблемы?

Спасибо заранее,

Answer 1

Размещая if more заявление в конце функции addCompaniesToIndex() вы практически сериализацию выполнения задачи: следующий отложенное задание не создаются, пока текущее отложенное задание не завершено индексированием своей доли Docs.

Что вы можете сделать, это переместить оператор if more сразу после вызова Company.query().fetch_page(), где вы получаете (большую часть) переменные, необходимые для выполнения следующей отложенной задачи.

Таким образом, следующая отложенная задача будет создана и завершена (длинная) до того, как будет завершена текущая, поэтому их обработка может быть перекрыта/пошатнулась. Вам также понадобятся некоторые другие модификации, например обработка переменной n_entities, которая теряет свое текущее значение в обновленном сценарии - но это более или менее косметическое/информационное, не существенное для фактической операции индексирования документа.

Если количество отложенных заданий очень велико, существует риск чередования многих из них одновременно, что может вызвать «взрыв» в количестве экземпляров, которые GAE создаст для их обработки. В таком случае вам не нужно «дросселировать» скорость, с которой отложенные задачи порождаются, задерживая их выполнение немного, см. https://stackoverflow.com/a/38958475/4495081.