如何优化Flask API快速发送PostgreSQL查询的7万条JSON数据

核心瓶颈分析 #

原方案的主要耗时点：

• pd.read_sql会把查询结果全量加载到DataFrame，额外占用内存且有转换开销
• jsonify基于标准JSON库，序列化大数据集效率偏低
• 未使用高效的数据库游标，内存占用过高
• 若查询用了SELECT *，会传输不必要的列数据

优化步骤及代码 #

1. 数据库层面优化（必做）

给查询用的日期字段加索引，减少数据库查询时间：

CREATE INDEX idx_your_table_date ON your_table(date_column);

查询时只选择需要的25列（不要用SELECT *），减少数据传输量：

SELECT col1, col2, ..., col25 FROM your_table WHERE date_column = %s;

2. 用psycopg2服务器端游标+ujson序列化（替代DataFrame中转）

使用psycopg2的namedtuple游标直接获取结构化数据，用ujson（比标准json快3-5倍）序列化，避免DataFrame的额外开销。

3. Flask响应直接返回序列化后的字符串

跳过jsonify，用flask.Response直接返回压缩后的JSON，减少框架层的处理开销。

完整优化代码 #

首先安装依赖：

pip install psycopg2-binary ujson flask flask-compress

数据库读取+API响应代码

import ujson
import psycopg2
from psycopg2.extras import NamedTupleCursor
from flask import Flask, Response, request
from flask_compress import Compress

app = Flask(__name__)
Compress(app)  # 开启响应压缩，减少传输体积

# 数据库连接配置（建议用连接池替代单次连接）
DB_CONFIG = {
    "dbname": "your_db",
    "user": "your_user",
    "password": "your_pwd",
    "host": "your_host"
}

@app.route('/get_daily_data', methods=['GET'])
def get_daily_data():
    target_date = request.args.get('date')
    if not target_date:
        return Response(ujson.dumps({"error": "Missing date parameter"}), 
                        mimetype='application/json', status=400)

    try:
        # 使用服务器端游标，避免一次性加载7万条数据到内存
        with psycopg2.connect(**DB_CONFIG) as conn:
            with conn.cursor(cursor_factory=NamedTupleCursor, name='server_side_cursor') as cur:
                # 替换为你实际需要的25列名
                query = """
                    SELECT col1, col2, col3, ..., col25 
                    FROM your_table 
                    WHERE date_column = %s
                """
                cur.execute(query, (target_date,))
                # 分批读取，降低内存压力
                batch_size = 10000
                results = []
                while True:
                    rows = cur.fetchmany(batch_size)
                    if not rows:
                        break
                    results.extend([row._asdict() for row in rows])

        # 用ujson快速序列化
        json_data = ujson.dumps(results)
        return Response(json_data, mimetype='application/json')

    except Exception as e:
        return Response(ujson.dumps({"error": str(e)}), 
                        mimetype='application/json', status=500)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

额外优化建议

• 使用数据库连接池（比如psycopg2.pool.SimpleConnectionPool），避免每次请求新建连接的开销
• 如果业务允许，考虑分页返回，让前端分批请求，减少单次响应的大小
• 若数据更新不频繁，可加入Redis缓存层，缓存每日查询结果，避免重复查询数据库

内容的提问来源于stack exchange，提问作者Lokesh Harad