Блог

Контекст

В панели мониторинга с использованием plotly Dash мне нужно выполнить дорогостоящую загрузку из БД только тогда, когда компонент (DataPicker с периодом для рассмотрения и, следовательно, для загрузки из БД) обновляется, а затем использовать полученный фрейм данных с другими компонентами (например, Выпадающие списки, фильтрующие фрейм данных), избегая дорогостоящего процесса загрузки.

В документах предлагается использовать dash_core_components.Store в качестве вывода обратный вызов, который возвращает кадр данных, сериализованный в json, а затем использовать Хранилище в качестве ввода других обратных вызовов, которые необходимо десериализовать из json в кадр данных.

Сериализация из/в JSON происходит медленно, и каждый раз, когда я обновляю компонент, требуется 30 секунд, чтобы обновить сюжет только для этого.

Я пытался использовать более быстрые сериализации, такие как рассол, паркет и перо, но в части десериализации я получаю сообщение об ошибке, указывающее, что объект пуст (при использовании JSON такая ошибка не появляется).

Вопрос

Можно ли выполнить сериализацию в хранилище Dash с помощью более быстрых методов, таких как рассол, перо или паркет (они занимают примерно половину времени для моего набора данных), чем JSON? Как?

Код

 import io
import traceback
import pandas as pd
from datetime import datetime, date, timedelta

import dash
import dash_core_components as dcc
import dash_html_components as html
import dash_bootstrap_components as dbc
from dash.dependencies import Input, Output
from plotly.subplots import make_subplots



app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
today = date.today()

app.layout = html.Div([
    dbc.Row(dbc.Col(html.H1('PMC'))),
    dbc.Row(dbc.Col(html.H5('analysis'))),
    html.Hr(),
    html.Br(),

    dbc.Container([
        dbc.Row([
            dbc.Col(
                dcc.DatePickerRange(
                    id='date_ranges',
                    start_date=today - timedelta(days=20),
                    end_date=today,
                    max_date_allowed=today, display_format='MMM Do, YY',
                ),
                width=4
            ),
        ]),
        dbc.Row(
            dbc.Col(
                dcc.Dropdown(
                    id='dd_ycolnames',
                    options=options,
                    value=default_options,
                    multi=True,
                ),
            ),
        ),
    ]),

    dbc.Row([
        dbc.Col(
            dcc.Graph(
                id='graph_subplots',
                figure={},
            ),
            width=12
        ),
    ]),

    dcc.Store(id='store')
])


@app.callback(
    Output('store', 'data'),
    [
        Input(component_id='date_ranges', component_property='start_date'),
        Input(component_id='date_ranges', component_property='end_date')
    ]
)
def load_dataset(date_ranges_start, date_ranges_end):
     # some expensive clean data step
     logger.info('loading dataset...')
     date_ranges1_start = datetime.strptime(date_ranges_start, '%Y-%m-%d')
     date_ranges1_end = datetime.strptime(date_ranges_end, '%Y-%m-%d')
     df = expensive_load_from_db(date_ranges1_start, date_ranges1_end)
     logger.info('dataset to json...')
     #return df.to_json(date_format='iso', orient='split')
     return df.to_parquet()                                 # <----------------------


@app.callback(
    Output(component_id='graph_subplots', component_property='figure'),
    [
        Input(component_id='store', component_property='data'),
        Input(component_id='dd_ycolnames', component_property='value'),
    ],
)
def update_plot(df_bin, y_colnames):
    logger.info('dataset from json')
    #df = pd.read_json(df_bin, orient='split')
    df = pd.read_parquet(io.BytesIO(df_bin))             # <----------------------
    logger.info('building plot...')
    traces = []
    for y_colname in y_colnames:
        if df[y_colname].dtype == 'bool':
            df[y_colname] = df[y_colname].astype('int')
        traces.append(
            {'x': df['date'], 'y': df[y_colname].values, 'name': y_colname},
        )
    fig = make_subplots(
        rows=len(y_colnames), cols=1, shared_xaxes=True, vertical_spacing=0.1
    )
    fig.layout.height = 1000
    for i, trace in enumerate(traces):
        fig.append_trace(trace, i 1, 1)
    logger.info('plotted')
    return fig


if __name__ == '__main__':
    app.run_server(host='localhost', debug=True)
 

Текст ошибки

OSError: Could not open parquet input source '<Buffer>': Invalid: Parquet file size is 0 bytes

В связи с обменом данными между клиентом и сервером, в настоящее время вы ограничены сериализацией JSON. Один из способов обойти это ограничение-использовать ServersideOutput компонент from dash-extensions , который хранит данные на сервере. По умолчанию он использует хранилище файлов и сериализацию pickle, но вы также можете использовать другие хранилища (например, Redis) и/или протоколы сериализации (например, arrow). Вот небольшой пример,

 import time
import dash_core_components as dcc
import dash_html_components as html
import plotly.express as px
from dash_extensions.enrich import Dash, Output, Input, State, ServersideOutput

app = Dash(prevent_initial_callbacks=True)
app.layout = html.Div([
    html.Button("Query data", id="btn"), dcc.Dropdown(id="dd"), dcc.Graph(id="graph"),
    dcc.Loading(dcc.Store(id='store'), fullscreen=True, type="dot")
])


@app.callback(ServersideOutput("store", "data"), Input("btn", "n_clicks"))
def query_data(n_clicks):
    time.sleep(1)
    return px.data.gapminder()  # no JSON serialization here


@app.callback(Input("store", "data"), Output("dd", "options"))
def update_dd(df):
    return [{"label": column, "value": column} for column in df["year"]]  # no JSON de-serialization here


@app.callback(Output("graph", "figure"), [Input("dd", "value"), State("store", "data")])
def update_graph(value, df):
    df = df.query("year == {}".format(value))  # no JSON de-serialization here
    return px.sunburst(df, path=['continent', 'country'], values='pop', color='lifeExp', hover_data=['iso_alpha'])


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