如何无需中间层每日更新Plotly Dash仪表盘的全局数据？

问题背景 #

我正在开发一个只读的Plotly Dash仪表盘，数据处理（数据库读取+转换）耗时约1分钟，因此把数据加载逻辑放在回调函数外，用全局变量存储，仅在首次启动时加载一次。当前数据为日粒度，需要每日自动更新全局数据，不想依赖Redis这类中间存储，该如何实现？

无中间层的解决方案 #

方法1：后台定时线程更新 #

利用Python的threading模块启动后台定时任务，每天自动调用数据加载函数更新全局数据。由于Dash是多线程运行，需加线程锁避免并发修改导致的数据不一致问题。

修改后的完整代码：

from dash import Dash, dcc, html, Input, Output, callback
import pandas as pd
from dash_ag_grid import AgGrid
import threading
import time
from datetime import datetime

# 全局变量与线程锁
global_data = pd.DataFrame()
data_lock = threading.Lock()
last_update_time = None

def get_global_data():
    data_df = pd.DataFrame()
    # 此处替换为你的数据库读取与转换逻辑
    # 示例模拟数据
    data_df = pd.DataFrame({
        'country': ['USA', 'USA', 'Canada', 'Canada', 'Mexico', 'Mexico'],
        'value': [100, 200, 150, 250, 80, 180]
    })
    return data_df

def update_global_data_daily():
    global global_data, last_update_time
    while True:
        with data_lock:
            print(f"正在更新全局数据：{datetime.now()}")
            global_data = get_global_data()
            last_update_time = datetime.now()
        # 等待24小时（86400秒），可按需调整间隔
        time.sleep(86400)

# 初始化全局数据
with data_lock:
    global_data = get_global_data()
    last_update_time = datetime.now()

# 启动定时更新线程（守护线程，随主进程退出）
update_thread = threading.Thread(target=update_global_data_daily, daemon=True)
update_thread.start()

app = Dash(__name__)

app.layout = html.Div([
    html.H6("选择国家查看数据！"),
    dcc.Dropdown(['USA', 'Canada', 'Mexico'], 'USA', id='country-dropdown'),
    html.Br(),
    html.Div(f"数据最后更新时间：{last_update_time.strftime('%Y-%m-%d %H:%M:%S')}", id='update-time'),
    AgGrid(
        id="data-grid",
        defaultColDef={"resizable": True, "sortable": True, "filter": True},
        columnSize="sizeToFit",
        style={'width': '100%'},
    )
])

def convert_df_to_dict(df):
    return df.to_dict('records')

def create_column_def(df):
    return [{'field': col} for col in df.columns]

@callback(
    Output('data-grid', 'rowData'),
    Output('data-grid', 'columnDefs'),
    Output('update-time', 'children'),
    Input(component_id='country-dropdown', component_property='value')
)
def update_table(input_country):
    with data_lock:
        filtered_data = global_data[global_data['country']==input_country]
        current_update_time = last_update_time.strftime('%Y-%m-%d %H:%M:%S')
    filtered_data_as_dict = convert_df_to_dict(filtered_data)
    column_defs = create_column_def(filtered_data)
    return filtered_data_as_dict, column_defs, f"数据最后更新时间：{current_update_time}"

if __name__ == '__main__':
    app.run(debug=False)  # 生产环境务必关闭debug模式，避免重复启动线程

方法2：本地文件缓存+系统定时任务 #

将处理好的全局数据保存为本地高效格式文件（比如Parquet），用系统定时任务（Linux的cron、Windows的任务计划程序）每天运行脚本重新生成该文件，Dash代码直接从本地文件读取数据。

步骤：

1. 编写独立脚本update_data.py，负责数据库读取、转换并保存数据：

import pandas as pd

def get_global_data():
    # 你的数据库读取与转换逻辑
    data_df = pd.DataFrame({
        'country': ['USA', 'USA', 'Canada', 'Canada', 'Mexico', 'Mexico'],
        'value': [100, 200, 150, 250, 80, 180]
    })
    return data_df

if __name__ == '__main__':
    df = get_global_data()
    df.to_parquet('global_data.parquet')

2. 配置系统定时任务，每天固定时间运行update_data.py
3. 修改Dash代码，从本地文件加载数据：

from dash import Dash, dcc, html, Input, Output, callback
import pandas as pd
from dash_ag_grid import AgGrid
import os
from datetime import datetime, timedelta

def load_global_data():
    try:
        return pd.read_parquet('global_data.parquet')
    except FileNotFoundError:
        # 文件不存在时初始化数据
        df = get_global_data()
        df.to_parquet('global_data.parquet')
        return df

global_data = load_global_data()

# 回调中可按需检查文件修改时间，重新加载数据
@callback(...)
def update_table(input_country):
    global global_data
    file_mtime = datetime.fromtimestamp(os.path.getmtime('global_data.parquet'))
    # 每5分钟检查一次文件是否更新
    if datetime.now() - file_mtime > timedelta(minutes=5):
        global_data = load_global_data()
    filtered_data = global_data[global_data['country']==input_country]
    # 后续处理逻辑...

方法3：回调触发时检查更新 #

每次用户交互触发回调时，检查当前时间与上次更新时间的间隔，若超过24小时则更新全局数据。该方案适合访问量较低的场景，避免无人访问时浪费资源。

核心修改逻辑：

from datetime import datetime, timedelta
import threading

global_data = get_global_data()
last_update_time = datetime.now()
data_lock = threading.Lock()

@callback(...)
def update_table(input_country):
    global global_data, last_update_time
    with data_lock:
        if datetime.now() - last_update_time > timedelta(hours=24):
            global_data = get_global_data()
            last_update_time = datetime.now()
        filtered_data = global_data[global_data['country']==input_country]
    # 后续处理逻辑...

注意事项 #

• 线程安全：使用多线程方案时，必须通过threading.Lock保护全局数据的读写操作
• 调试模式：Dash的debug模式会自动重载代码，可能导致定时线程重复启动，生产环境需关闭debug
• 异常处理：在get_global_data()中添加异常捕获，避免更新失败导致全局数据丢失，可保留旧数据或记录日志

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