Python 3.6 asyncio - исключение задачи не было получено - задание получило плохой выход: 200 - PullRequest
Новая
       12

Python 3.6 asyncio - исключение задачи не было получено - задание получило плохой выход: 200

3 голосов
/ 23 марта 2019

Я прочитал другие вопросы и ответы и до сих пор не могу понять, что я делаю здесь не так.

Я пытаюсь создать производителя Elasticsearch 6.x в Python 3.6, используя асинхронную реализацию ES (https://github.com/elastic/elasticsearch-py-async), и пока он работает (записи успешно помещаются в ES), я получаю *Ошибки 1005 * и * 1006. * Я полагаю, что они оба являются результатом одной и той же проблемы, и одна, вероятно, вызывает другую?

Я использую следующие модули: 

python 3.6
elasticsearch=6.3.1
elasticsearch-async=6.2.0
boto3=1.9.118

Нижемой код: 

import json
import boto3
import logging
import os
import gzip
import asyncio
from elasticsearch import RequestsHttpConnection
from elasticsearch_async import AsyncElasticsearch
from assume_role_aws4auth import AssumeRoleAWS4Auth
import time

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Operating constants
MAX_RECORDS_IN_BATCH = 500
MAX_BATCH_SIZE = 10000000

# boto3 clients
credentials = boto3.Session().get_credentials()
awsauth = AssumeRoleAWS4Auth(credentials, 'us-east-1', 'es')

cloudwatch_client = boto3.client('cloudwatch')
s3_resource = boto3.resource('s3')
event_loop = asyncio.get_event_loop()

es_client = AsyncElasticsearch(hosts=['https://ES_HOST'], http_compress=True, http_auth=awsauth, use_ssl=True,
                               verify_certs=True, connection_class=RequestsHttpConnection, loop=event_loop)


def lambda_handler(filename, context):
    event_loop.run_until_complete(process(filename))
    pending = asyncio.Task.all_tasks()
    event_loop.run_until_complete(asyncio.gather(*pending))


async def process(filename: str):
    for action_chunk in read_chunk(filename, MAX_BATCH_SIZE, MAX_RECORDS_IN_BATCH):
        try:
            resp = asyncio.ensure_future(es_client.bulk(body=action_chunk, index='index', doc_type='type', _source=False))
            await asyncio.sleep(.1)
        except Exception as ex:
            logger.error(ex)


def read_chunk(file_path: str, max_batch_size: int, max_records: int):
    actions: str = ''
    actions_size: int = 0
    num_actions: int = 0
    with gzip.open(file_path, 'rt') as f:
        for line in f:
            request = json.dumps(dict({'index': dict({})})) + '\n' + line + '\n'
            request_size = len(request.encode('utf-8'))

            # Check to see if this record will put us over the limits
            if (actions_size + request_size) > max_batch_size or num_actions == max_records:
                yield actions
                actions = ''
                num_actions = 0
                actions_size = 0

            # Add the record
            actions += request
            num_actions += 1
            actions_size += request_size

    if actions != '':
        yield actions


if __name__ == '__main__':
    lambda_handler('/path/to/file', None)

Ниже приведена ошибка, которую я получаю каждый раз, когда звоню es_client.bulk: 

Task exception was never retrieved
future: <Task finished coro=<AsyncTransport.main_loop() done, defined at /path/to/PythonElasticsearchIngest/venv/lib/python3.6/site-packages/elasticsearch_async/transport.py:143> exception=RuntimeError('Task got bad yield: 200',)>
Traceback (most recent call last):
  File "/path/to/PythonElasticsearchIngest/venv/lib/python3.6/site-packages/elasticsearch_async/transport.py", line 150, in main_loop
    method, url, params, body, headers=headers, ignore=ignore, timeout=timeout)

Может кто-нибудь сказать мне, что я здесь делаю неправильно? Кроме того,если есть что-то, что я могу сделать лучше / эффективнее, я бы хотел это услышать. Я хотел использовать пакет Helpers, но его реализация asyncio отсутствует.

1 Ответ

1 голос
/ 23 марта 2019

Я не уверен, что это проблема, но вот что может случиться.

Вы создаете несколько задач внутри process() сопрограммы, но не сохраняете ссылки на них.Это может привести к проблеме: некоторые задачи собирают мусор , прежде чем вы сможете явно получить их результаты.Если такое случается asyncio предупреждает вас о ситуации.

Для ее решения необходимо сохранить все созданные задачи и убедиться, что все они ожидаются: 

tasks = []

# ...

async def process(filename: str):
    # ...
    task = asyncio.ensure_future(...)
    tasks.append(task)
    # ...


def lambda_handler(filename, context):
    # ...
    event_loop.run_until_complete(asyncio.gather(*tasks ))

Если мои предположения верны, вы, вероятно, увидите, что RuntimeError('Task got bad yield: 200',) поднят на lambda_handler.Вы можете получить все исключения, не поднимая их, передав параметр return_exceptions=True в asyncio.gather .Таким образом, вы избегаете предупреждений (но не основную причину, по которой эти исключения произошли, это прежде всего задачи).

Извините, не могу помочь дальше, чем здесь.

Upd:

Я изменил ответ, исправляя ошибку оригинальной версии.

...