如何在不将数据保存到 csv 文件的情况下使用 FastAPI 流式传输 DataFrame？

Question

我想知道如何使用 FastAPI 流式传输 DataFrame 而无需将 DataFrame 保存到磁盘上的 csv 文件中。目前，我设法做的是从 csv 文件中流式传输数据，但与返回 FileResponse 相比，速度不是很快。下面的/option7 是我想要做的。

我的目标是从 FastAPI 后端流式传输数据，而不将 DataFrame 保存到 csv 文件。

谢谢你。

from fastapi import FastAPI, Response,Query
from fastapi.responses import FileResponse,HTMLResponse,StreamingResponse
app = FastAPI()

df = pd.read_csv("data.csv")

@app.get("/option4")
def load_questions():
    return FileResponse(path="C:Downloads/data.csv", filename="data.csv")

@app.get("/option5")
def load_questions():
    def iterfile():  # 
        with open('data.csv', mode="rb") as file_like:  # 
            yield from file_like  # 

    return StreamingResponse(iterfile(), media_type="text/csv")

@app.get("/option7")
def load_questions():
    def iterfile():  # 
        #with open(df, mode="rb") as file_like:  # 
        yield from df  # 

    return StreamingResponse(iterfile(), media_type="application/json")

Answer 1

推荐方法

正如this answer 以及here 和here 中所提到的，当整个数据（或在您的情况下为DataFrame）已经加载到内存中时，使用StreamingResponse 没有什么意义。 StreamingResponse 当您想要传输实时数据并且您不提前知道输出的大小并且不想等到收集所有数据后再开始发送数据时才有意义到客户端，以及当您要返回的文件太大而无法放入内存时（例如，如果您有 8GB 的​​ RAM，则无法加载 50GB 的文件），因此，您宁愿加载文件分块存入内存。

在您的情况下，由于 DataFrame 已经加载到内存中，您应该在使用 Panda 的 .to_json() 方法将 DataFrame 转换为 JSON 字符串后直接返回自定义 Response，如 this answer 中所述（请参阅这个related post 也是如此）。例子：

from fastapi import Response

@app.get("/")
def main():
    return Response(df.to_json(orient="records"), media_type="application/json")

如果您发现浏览器需要一段时间才能显示数据，您可能需要下载数据作为.json 文件发送到用户的设备（这将更快地完成），而不是等待浏览器显示大量数据。您可以通过使用attachment 参数在Response 中设置Content-Disposition 标头来做到这一点（有关详细信息，请参阅this answer）：

@app.get("/")
def main():
    headers = {'Content-Disposition': 'attachment; filename="data.json"'}
    return Response(df.to_json(orient="records"), headers=headers, media_type='application/json')

您还可以使用.to_csv() 方法将数据作为.csv 文件返回，而无需指定路径参数。由于使用return df.to_csv() 会导致在浏览器中显示包含 字符的数据，您可能会发现最好将csv 数据放在Response 中，并指定Content-Disposition 标头，以便数据将作为.csv 文件下载。例子：

@app.get("/")
def main():
    headers = {'Content-Disposition': 'attachment; filename="data.csv"'}
    return Response(df.to_csv(), headers=headers, media_type="text/csv")

不推荐的方法

使用StreamingResponse（您不应该，如前所述），您需要使用标准的json 库或其他更快的 JSON 编码器，如@987654331 中所述，遍历 DataFrame 中的行，将每一行转换为字典，然后转换为 JSON 字符串@（JSON 字符串稍后将在 FastAPI 内部编码为 byte 格式）。例子：

@app.get("/")
def main():
    def iter_df():
        for _, row in df.iterrows():
            yield json.dumps(row.to_dict()) + '
'

    return StreamingResponse(iter_df(), media_type="application/json")

遍历 pandas 对象通常很慢，并且不建议.如this answer 中所述：

Pandas 中的迭代是一种反模式你应该 仅在您用尽所有其他选项时才这样做.你应该 不是使用名称中带有"iter" 的任何函数不止几次 千行，否则您将不得不习惯很多的等待。

更新

正如@Panagiotis Kanavos 在下面的 cmets 部分中指出的那样，在已加载到内存中的 DataFrame 上使用 .to_json() 或 .to_csv() 会导致将整个输出字符串分配到内存中，从而使 RAM 使用量翻倍甚至更糟。因此，如果使用上述任何一种方法，如果有如此大量的数据可能导致系统变慢或崩溃（因为内存不足），您应该使用StreamingResponse，如前所述。您可以在this post 中找到更快的iterrows() 替代方法，以及更快的JSON 编码器，例如orjson 和ujson，如this answer 中所述。

或者，您可以将数据保存到磁盘，然后删除 DataFrame 以释放内存——您甚至可以使用gc.collect() 手动触发垃圾收集，如this answer 所示；但是，不鼓励频繁调用垃圾收集，因为这是一项代价高昂的操作并且可能会影响性能——并返回 FileResponse（假设数据可以放入 RAM；否则，您应该使用 StreamingResponse，参见 this answer好吧），最后，有一个BackgroundTask 在返回响应后从磁盘中删除文件。下面给出示例。

无论如何，您可以选择的解决方案应该基于您的应用程序的要求，例如，您希望同时服务的用户数量、数据大小、响应时间等），以及您的系统规格（例如，可用内存分配）。此外，由于对 panda 的 DataFrame 方法的所有调用都是同步的，因此您应该记住使用普通的def 定义您的端点，以便它在外部线程池中运行；否则，它将阻止服务器。或者，您可以使用来自concurrency 模块的Starlette 的run_in_threadpool()，它将在单独的线程中运行to_csv() 或to_json() 函数，以确保主线程（运行协程的位置）不会被阻塞。有关def 与async def 的更多详细信息，请查看this answer。

from fastapi import BackgroundTasks
from fastapi.responses import FileResponse 
import uuid
import os

@app.get("/")
def main(background_tasks: BackgroundTasks):
    filename = str(uuid.uuid4()) + ".csv"
    df.to_csv(filename)
    del df  # release the memory
    background_tasks.add_task(os.remove, filename) 
    return FileResponse(filename, filename="data.csv", media_type="text/csv")