Flask RESTful API返回超23万行JSON响应过慢的解决方案问询

可行优化方案，按落地优先级排序： #

1. 全局预加载静态数据，消除每次请求的文件IO耗时
   你当前代码每次收到请求都会重新读取本地CSV文件，23万行的文件IO开销是响应慢的核心原因之一。可以在服务启动时就把CSV加载到内存，提前预处理好中英文两种列名的数据集，每次请求直接用内存里的数据过滤，不用重复读盘。
2. 开启响应Gzip压缩
   120MB的纯JSON文本开启gzip压缩后体积会降到10~15MB左右，不管是Postman还是Power BI拉取都不会有大小限制问题，Flask可以直接用flask-compress库一键开启，不需要改业务逻辑。
3. 优化JSON序列化逻辑，避免冗余内存占用
   你现在的逻辑是先把dataframe转成Python字典，再由Flask序列化成JSON，中间多了一层大字典的内存开销，直接用pandas内置的to_json()方法生成JSON字符串返回，或者改用流式响应分批输出，内存占用能降70%以上，序列化速度也快很多。
4. 可选优化：新增分页/CSV导出能力
   如果后续数据量还会上涨，可以加limit、offset参数支持分页拉取；也可以新增format参数，支持直接返回CSV格式响应，BI工具拉取CSV的适配性比大JSON好很多，相同数据量的CSV体积只有JSON的60%左右。
5. 可选优化：数据转存轻量数据库
   如果过滤维度后续会增加，可以把CSV数据导入SQLite数据库，过滤的时候直接走SQL查询，比pandas的loc过滤效率更高，内存占用也更低。

参考修改后的代码示例 #

# app.py
from flask import Flask, make_response
from flask_restful import Resource, Api, reqparse
import pandas as pd
import os
import pathlib
from flask_compress import Compress

# 启动Flask应用
app = Flask(__name__)
# 开启gzip压缩，默认对json等文本类型自动压缩
Compress(app)
api = Api(app, default_mediatype='application/json')

# 服务启动时预加载数据，只执行一次
BASE_PATH = pathlib.Path(__file__).parent.resolve()
CSV_PATH = os.path.join(BASE_PATH, "static", "csv", "DimensionCalendar.csv")
# 预加载原始数据
df_origin = pd.read_csv(CSV_PATH)
# 预加载葡萄牙语列名的数据集，避免每次请求重复改列名
pt_columns = ['DT_DATA', 'NR_ANO', 'NR_MES', 'NR_DIA', 'NM_MES', 'NM_DIADASEMANA', 'NR_MESANO', 'NM_MESANO', 'NM_DATALONGA', 'NR_DIADASEMANA', 'NR_DIADOANO', 'NR_SEMESTRE', 'NM_SEMESTRE', 'NR_QUADRIMESTRE', 'NR_TRIMESTRE', 'NM_TRIMESTRE', 'NR_BIMESTRE', 'NM_BIMESTRE', 'NR_ANORELATIVO', 'NR_MESRELATIVO', 'NR_DIARELATIVO', 'NR_DATA', 'NR_ANOMES', 'NR_ANOTRIMESTRE', 'IC_DIAUTIL', 'IC_FINALDESEMANA', 'IC_FERIADO', 'IC_ANOATUAL', 'IC_MESATUAL', 'IC_DATAATUAL', 'IC_ANOPASSADO', 'IC_MESPASSADO', 'IC_ONTEM', 'LINDATA', 'LINORIGEM']
df_pt = df_origin.copy()
df_pt.columns = pt_columns

# 首页API类
class Index(Resource):
    def get(self):
        index_data = [
        {
            "href": "/dimensioncalendar",
            "name": "Dimension Calendar",
            "description": "带日期属性的日历维度接口"
        }
        ]
        return index_data, 200

# 日历维度API类
class DimensionCalendar(Resource):
    def get(self):
        parser = reqparse.RequestParser()
        parser.add_argument('start_dt')
        parser.add_argument('end_dt')
        parser.add_argument('lang')
        args = parser.parse_args()

        # 直接用预加载的数据集，不用重复读CSV
        if args.get('lang') == 'pt-BR':
            data = df_pt.copy()
            date_col = 'DT_DATA'
        else:
            data = df_origin.copy()
            date_col = 'DT_SHORTDATE'

        # 日期过滤
        if args['start_dt'] and args['end_dt']:
            data = data.loc[(data[date_col] >= str(args['start_dt'])) & (data[date_col] <= str(args['end_dt']))]
        elif args['start_dt'] and not args['end_dt']:
            data = data.loc[data[date_col] == str(args['start_dt'])]

        # 直接生成JSON字符串返回，避免转字典的冗余开销
        # 如果需要保持原有接口的列导向JSON结构，把orient参数改成'dict'即可
        resp = make_response(data.to_json(orient='records'))
        resp.headers['Content-Type'] = 'application/json'
        return resp

# 404处理
@app.errorhandler(404)
def page_not_found(e):
    return "<h1>404</h1><p>资源不存在</p>", 404

# 注册路由
api.add_resource(DimensionCalendar, '/dimensioncalendar')
api.add_resource(Index, '/')

if __name__ == '__main__':
    app.run()

内容的提问来源于stack exchange，提问作者Guilherme Matheus