Plotly Dash多实例下DynamoDB数据定时更新及加载消息实现问询

一、独立后端进程处理数据拉取与清洗 #

核心思路是把数据拉取、清洗逻辑从Dash应用中剥离，用单一定时进程执行，结果存到共享存储（如AWS S3、EFS），所有Dash实例只读取共享数据。

实现步骤： #

1. 编写独立数据处理脚本
   封装你的数据拉取和清洗逻辑，最后将结果保存到共享存储：

   import pandas as pd
   import boto3
   from your_modules import FetchData, CleanData  # 导入你的自定义模块
   
   def process_and_save_data():
       try:
           # 拉取原始数据
           FetchData("Sessions2").scan_save_all_records()
           # 清洗数据
           clean_df = CleanData.clean_save_raw_data()
           # 保存到S3（替换为你的桶和路径）
           s3_client = boto3.client("s3")
           s3_client.put_object(
               Bucket="your-data-bucket",
               Key="clean_data/latest_data.csv",
               Body=clean_df.to_csv(index=False).encode("utf-8")
           )
           print("数据更新完成")
       except Exception as e:
           print(f"数据处理失败：{str(e)}")
           # 可选：添加告警逻辑（如发送邮件/CloudWatch告警）
   

2. 定时触发脚本
   选择以下一种方式实现每10分钟执行一次：

   • APScheduler（Python定时库）：适合长期运行的进程

     from apscheduler.schedulers.blocking import BlockingScheduler
     
     if __name__ == "__main__":
         scheduler = BlockingScheduler()
         # 每10分钟执行一次
         scheduler.add_job(process_and_save_data, "interval", minutes=10)
         scheduler.start()
     

     部署时可将脚本作为systemd服务运行，确保进程稳定。
   • AWS Lambda + CloudWatch Events：Serverless方案，无需维护服务器
     把上述处理逻辑封装成Lambda函数，用CloudWatch Events设置每10分钟触发一次，权限配置需包含DynamoDB读取和S3写入。

3. 避免重复执行
   若数据处理耗时超过10分钟，可添加锁机制防止并发执行：

   • 用DynamoDB创建一个锁表，脚本执行前先获取锁，执行完成后释放；
   • 用文件锁（仅适用于同一服务器/共享目录的场景）。

二、Dash应用实例读取共享数据 #

修改Dash应用的回调逻辑，不再直接调用DynamoDB，而是读取共享存储中的CSV文件。

代码示例： #

import dash
from dash import dcc, html, Input, Output, dash_table
import pandas as pd
import boto3

app = dash.Dash(__name__)

app.layout = html.Div([
    # 每10分钟触发一次数据更新（与后端进程同步）
    dcc.Interval(
        id="data-update-interval",
        interval=10*60*1000,  # 10分钟（毫秒）
        n_intervals=0
    ),
    # 加载提示组件
    dcc.Loading(
        id="loading-container",
        children=[html.Div(id="data-display-area")],
        type="circle",  # 可选加载动画类型：circle/bar/dot
        fullscreen=False
    )
])

@callback(
    Output("data-display-area", "children"),
    Input("data-update-interval", "n_intervals"),
    prevent_initial_call=False
)
def update_data_display(n):
    # 从S3读取清洗后的CSV
    s3_client = boto3.client("s3")
    obj = s3_client.get_object(
        Bucket="your-data-bucket",
        Key="clean_data/latest_data.csv"
    )
    df = pd.read_csv(obj["Body"])

    # 生成可视化内容（示例：DataTable）
    return dash_table.DataTable(
        data=df.to_dict("records"),
        columns=[{"name": col, "id": col} for col in df.columns],
        page_size=20,
        style_table={"overflowX": "auto"}
    )

if __name__ == "__main__":
    app.run_server(debug=True)

三、自定义中间加载提示 #

除了dcc.Loading的默认动画，还可以实现自定义文本提示：

代码示例： #

# 修改布局，添加加载文本
app.layout = html.Div([
    dcc.Interval(id="data-update-interval", interval=10*60*1000, n_intervals=0),
    html.Div(id="loading-status", style={"color": "#666", "margin": "10px 0"}),
    html.Div(id="data-display-area")
])

@callback(
    Output("data-display-area", "children"),
    Output("loading-status", "children"),
    Input("data-update-interval", "n_intervals"),
    prevent_initial_call=False
)
def update_data_display(n):
    # 先返回加载提示
    loading_msg = "正在获取最新数据，请稍候..."
    # 读取数据
    s3_client = boto3.client("s3")
    obj = s3_client.get_object(Bucket="your-data-bucket", Key="clean_data/latest_data.csv")
    df = pd.read_csv(obj["Body"])
    # 生成表格
    table = dash_table.DataTable(
        data=df.to_dict("records"),
        columns=[{"name": col, "id": col} for col in df.columns]
    )
    # 加载完成，清空提示
    return table, ""

四、部署关键注意事项 #

• 共享存储权限：确保Dash实例和后端进程都拥有共享存储的读写权限（如S3桶的IAM权限、EFS的挂载权限）；
• 错误处理：在后端脚本和Dash回调中添加异常捕获，避免进程崩溃或页面报错；
• 缓存优化：可使用Flask-Caching对读取的CSV数据进行缓存，减少重复读取共享存储的开销。

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