Python 3.6 asyncio - 从未检索到任务异常 - 任务 yield 不佳 : 200

Question

我已经阅读了其他问题和答案，但仍然无法弄清楚我在这里做错了什么。

我正在尝试使用 ES 的异步实现 (https://github.com/elastic/elasticsearch-py-async) 在 Python 3.6 中创建一个 Elasticsearch 6.x 生产者，当它工作时(记录已成功推送到 ES)，我得到 Task Exception was从未重试 并且 Task got bad yield: 200 错误。我假设它们都是由同一个问题引起的，一个可能导致另一个？

我正在使用以下模块:

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 实现。

Answer 1

我不确定这是否是问题所在，但可能会发生以下情况。

您在 process() 协程中创建多个任务，但不存储对它们的引用。可能会导致一个问题:有些任务是garbage collected在您可以显式检索他们的结果之前。如果发生这样的事情 asyncio warns你的情况。

要解决它，您应该存储所有已创建的任务并确保所有这些任务都已等待:

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 .这样你就可以避免警告(但不是那些异常发生的根本原因，它首先是任务)。

抱歉，不能提供更多帮助。

更新:

我修改了原版答案修复错误。