Python 3.6 asyncio - 从未检索到任务异常 - 任务产量不佳：200

Question

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

我正在尝试使用 ES 的 asyncio 实现（https://github.com/elastic/elasticsearch-py-async）在 Python 3.6 中创建一个 Elasticsearch 6.x 生产者，并且在它工作时（记录已成功推送到 ES） ，我得到Task Exception was never retried和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()，但不存储对它们的引用。这可能会导致一个问题：某些任务在您可以显式检索其结果之前就被垃圾收集了。如果发生这种情况，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。这样您就可以避免警告（但不是这些异常发生的根本原因）。

抱歉，无法提供比这里更多的帮助。

更新：

我更改了答案，修复了原始版本的错误。