如何让Cloud Run上的Dash应用动态读取Google Cloud Storage数据？

方案1：按需加载数据（每次回调/请求时读取） #

• 核心逻辑：不在应用启动时一次性把数据加载到内存，而是在每次Dash回调触发时，直接从GCS拉取最新数据。
• 优势：确保每次使用的都是GCS上的最新数据，彻底规避内存缓存旧数据的问题。
• 优化建议：如果数据体积大或读取耗时久，可搭配带过期时间的缓存（比如用cachetools.TTLCache设置7天过期），减少重复读取GCS的次数。
• 示例代码：

import dash
from dash import dcc, html, Input, Output
from google.cloud import storage
import pandas as pd
from cachetools import TTLCache

# 缓存配置：最多存1条数据，7天后自动过期
cache = TTLCache(maxsize=1, ttl=60*60*24*7)

def get_latest_data():
    if 'data' not in cache:
        # 从GCS读取并处理数据
        client = storage.Client()
        bucket = client.get_bucket("your-bucket-name")
        blob = bucket.blob("your-data-file.csv")
        df = pd.read_csv(blob.open())
        cache['data'] = df
    return cache['data']

app = dash.Dash(__name__)

app.layout = html.Div([
    dcc.Graph(id="data-visualization"),
    # 可选：定时触发回调刷新，比如每小时一次
    dcc.Interval(id="refresh-interval", interval=3600000, n_intervals=0)
])

@app.callback(
    Output("data-visualization", "figure"),
    Input("refresh-interval", "n_intervals")
)
def update_visualization(_):
    df = get_latest_data()
    # 生成可视化图表
    return {
        "data": [{"x": df["date"], "y": df["value"], "type": "line"}],
        "layout": {"title": "Weekly Updated Data"}
    }

if __name__ == "__main__":
    app.run_server(host="0.0.0.0", port=8080)

方案2：后台线程定时刷新内存数据 #

• 核心逻辑：应用启动时加载初始数据，同时启动后台线程，每周自动从GCS拉取最新数据并更新内存中的全局变量。
• 优势：回调时直接读取内存数据，响应速度快；无需手动干预，自动完成数据刷新。
• 注意事项：必须加线程锁避免并发读写导致的数据异常；Cloud Run实例缩容重启后，会自动重新加载最新数据。
• 示例代码：

import dash
from dash import dcc, html, Input, Output
from google.cloud import storage
import pandas as pd
import threading
import time

# 全局变量存储数据，搭配锁保证线程安全
current_data = None
data_lock = threading.Lock()

def refresh_data():
    global current_data
    with data_lock:
        # 从GCS拉取最新数据
        client = storage.Client()
        bucket = client.get_bucket("your-bucket-name")
        blob = bucket.blob("your-data-file.csv")
        current_data = pd.read_csv(blob.open())
    # 每周执行一次刷新（60*60*24*7秒）
    threading.Timer(60*60*24*7, refresh_data).start()

# 初始化加载数据并启动定时线程
refresh_data()

app = dash.Dash(__name__)

app.layout = html.Div([
    dcc.Graph(id="data-visualization")
])

@app.callback(
    Output("data-visualization", "figure"),
    Input("data-visualization", "id")  # 页面加载时触发回调
)
def update_visualization(_):
    with data_lock:
        df = current_data.copy()
    # 生成可视化图表
    return {
        "data": [{"x": df["date"], "y": df["value"], "type": "bar"}],
        "layout": {"title": "Cached Weekly Data"}
    }

if __name__ == "__main__":
    app.run_server(host="0.0.0.0", port=8080)

方案3：触发式数据刷新（通过专用端点） #

• 核心逻辑：在Dash应用中添加一个API端点，当GCS数据更新后（比如通过Cloud Function监听GCS变更事件），调用该端点触发数据刷新。
• 优势：精准控制刷新时机，避免不必要的重复读取操作。
• 示例代码（结合Flask路由）：

import dash
from dash import dcc, html, Input, Output
from google.cloud import storage
import pandas as pd
from flask import Flask, request

server = Flask(__name__)
app = dash.Dash(__name__, server=server)

current_data = None

def load_latest_data():
    global current_data
    client = storage.Client()
    bucket = client.get_bucket("your-bucket-name")
    blob = bucket.blob("your-data-file.csv")
    current_data = pd.read_csv(blob.open())

# 初始化加载数据
load_latest_data()

# 添加刷新端点
@server.route("/refresh-data", methods=["POST"])
def refresh_data():
    load_latest_data()
    return {"status": "success", "message": "Data refreshed"}

app.layout = html.Div([
    dcc.Graph(id="data-visualization")
])

@app.callback(
    Output("data-visualization", "figure"),
    Input("data-visualization", "id")
)
def update_visualization(_):
    df = current_data
    return {
        "data": [{"x": df["date"], "y": df["value"], "type": "line"}],
        "layout": {"title": "Trigger-Refreshed Data"}
    }

if __name__ == "__main__":
    app.run_server(host="0.0.0.0", port=8080)

• 配套操作：用Cloud Function监听GCS的对象更新事件，当目标数据文件更新时，自动发送POST请求到/refresh-data端点触发刷新。

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