如何利用流式传输优化异步数据保存？

2024-03-28 23:53:34

利用流式传输优化异步数据保存

在编写代码时，我们通常会使用 await 来等待异步操作完成，例如保存数据到数据库。然而，在处理大数据集时，这种方法会导致内存使用激增，因为我们需要等待所有数据收集完成后再进行保存。

流式传输的优势

为了解决这个问题，我们可以利用流式传输来异步地保存数据。流式传输允许我们逐块处理数据，并在数据生成时将其保存到数据库中。这样，我们就避免了在内存中累积大量数据，从而减少了内存使用。

优化后的代码

以下是使用流式传输优化异步数据保存的优化代码：

async def get_transactions_and_save_to_land(api, practice_ids, table_name, conn_object, start_year, end_year, incremental, with_deleted, date_modified, batch_size=100000):

    # for concurrent api requests
    sem1 = asyncio.Semaphore(4)

    if not incremental:
        get_tasks = []
        for practice_id, year in itertools.product(practice_ids, range(start_year, end_year + 1)):
            task = asyncio.create_task(get_transactions(api, practice_id, year, incremental=incremental, with_deleted=with_deleted, date_modified=date_modified, semaphore=sem1))
            get_tasks.append(task)
        
        results = await asyncio.gather(*get_tasks) 

    reader = pd.read_csv(io.BytesIO(), chunksize=1000)

    for transactions in results:
        for chunk in pd.read_csv(io.StringIO(transactions), chunksize=1000):
            chunk.to_sql(table_name, conn_object, if_exists='append', index=False)