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

Posted 2023-02-16

【中文标题】Python 3.6 asyncio - 从未检索到任务异常 - 任务的产量不好：200

【英文标题】：Python 3.6 asyncio - Task exception was never retrieved - Task got bad yield: 200

【发布时间】：2019-03-23 16:05:11

【问题描述】：

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

我正在尝试使用 ES 的 asyncio 实现 (https://github.com/elastic/elasticsearch-py-async) 在 Python 3.6 中创建一个 Elasticsearch 6.x 生产者，虽然它可以工作（记录已成功推送到 ES），但我得到了 Task Exception was never retried 和 @987654323 @错误。我认为它们都是由同一个问题引起的，而一个可能导致另一个？

我正在使用以下模块：

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】：

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

您在 process() 协程中创建多个任务，但不存储对它们的引用。这可能会导致一个问题：某些任务是garbage collected，然后您才能显式检索它们的结果。如果发生这种情况asynciowarns你关于情况。

要解决这个问题，您应该存储所有创建的任务并确保所有任务都处于等待状态：

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。这样您就可以避免警告（但不是这些异常发生的根本原因）。

抱歉，这里帮不上忙。

更新：

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

【讨论】：

我想知道是否有必要在run_until_complete(gather(...))之后添加task.result()。 gather 和 run_until_complete 不应该已经访问了任务结果以返回它们吗？

那么“坏产量”是我应该能够回溯的潜在错误吗？

@user4815162342 你说得对，我把gather 的行为和wait 的行为搞混了。谢谢你纠正我！我用另一个假设改变了答案。

@***s 完全正确。它可能发生在es_client.bulk 内部的某个地方。记下，我关于警告原因的版本是错误的，我将其更改为（我希望）更正。

好主意，GC 在代码之前完成任务很可能是问题所在。我仍然想知道在@***s 的代码代码中对run_until_complete 的第二次调用 - 是否有必要以及它应该完成什么，因为它看起来很危险，因为很难判断将返回哪些任务。 @***s，你能评论那部分吗？