Python - 将数据从文件流式传输到启用异步的 Kinesis Producer

Question

我正在使用 Python 3.6 中启用异步的 Kinesis Producer 模块，以部署到 AWS Lambda（因此我需要它与 3.6 兼容）。

我的用例是从磁盘延迟读取文件（大约 100MB 压缩 - 1GB 未压缩）并将数据（一次 500 行）流式传输到 Kinesis Producer。我希望 Kinesis Producer 在我读取下一批 500 行时开始将 500 条记录推送到 Kinesis。

我注意到的是，它一次读取整个文件 500 行，然后开始将数据推送到 Kinesis Producer。原因似乎是因为我没有打电话给await asyncio.sleep(1)，但我也不知道我这样做是否正确。

def lambda_handler(event, context):
    event_loop = asyncio.get_event_loop()
    # Extract filename from event and download file from S3
    event_loop.run_until_complete(process(filename))
    pending = asyncio.Task.all_tasks()
    event_loop.run_until_complete(asyncio.gather(*pending))


async def process(filename):
    for chunk in read_lines(filename, MAX_RECORDS_IN_BATCH):
    asyncio.ensure_future(write_kinesis(chunk)).add_done_callback(callback)


def callback(result):
    print(str(result))


async def write_kinesis(records):
    future = asyncio.ensure_future(producer.put_records(records=records))

如果我将await asyncio.sleep(.1) 添加到process(filename) 函数的末尾，它似乎完全符合我的要求，当然，它实际上会阻塞主线程0.1 秒。

Q- 使用 asyncio.sleep 阻塞足够长的时间让 Kinesis Producer 将数据推送出去，这就是诀窍吗？它睡得越少，我在内存中保存的数据就越多，因为 kinesis 客户端没有太多时间将数据推出，但它会运行得更快（在一定程度上）？

Q- 我这样做是否正确？同样，我尝试读取 500 行，推送到 kinesis（异步），在 Kinesis 客户端工作时再读取 500 行，冲洗并重复。

另外，当从回调函数中查看打印语句时，我注意到如果 write_kinesis 函数没有返回任何内容，则回调的 print 语句有result=None，而如果 write_kinesis 函数返回 Future，则回调的 print 语句有result=<Task pending...11b35f18>()。

Q-我假设没有return语句没有结果，但是为什么在状态仍然是“Pending”时调用回调函数？

编辑 1：我忘了说，Kinesis 客户端已经启用了异步功能。

Answer 1

你只需用 await 调用异步函数

async def process(filename):
    for chunk in read_lines(filename, MAX_RECORDS_IN_BATCH):
        await producer.put_records(records=chunk)

async def run():
    # Extract filename from event and download file from S3
    await process(filename)

loop = asyncio.get_event_loop()
loop.run_until_complete( run() )
loop.close()