python - グラフの計算中は Das ブロードキャストを使用できません

Question

私は Dask を試していて、pandas.DataFrameすべてのワーカー ノードにルックアップを送信したいと考えています。残念ながら、次のように失敗します。

TypeError: ("'Future' object is not subscriptable", 'occurred at index 0')

lookup['baz'].iloc[2]を使用する代わりに、lookup.result()['foo'].iloc[2]正常に動作しますが、入力データフレームのインスタンスが大きい場合、何度もスタックしているようfrom_pandasです。また、将来を手動でブロックする必要があるのは奇妙に思えます (適用操作の行ごとに何度も何度も。ワーカー ノードごとに 1 回だけ将来をブロックする方法はありますか?単純な改善は を使用することである可能性がありますmap_partitionsが、これは、パーティションの数がかなり少ない場合にのみ実行可能です。

import pandas as pd
import dask.dataframe as dd
from dask.distributed import Client

client = Client()

df_first = pd.DataFrame({'foo':[1,2,3]})
df_second = pd.DataFrame({'bar':[1,2,3], 'baz':[1,2,3]})

df_first_scattered = client.scatter(df_first, broadcast=True)
df_second_dask = dd.from_pandas(df_second, npartitions=2)


def foo(row, lookup):
    # TODO some computation which relies on the lookup
    return lookup['foo'].iloc[2]

df_second_dask['foo'] = df_second_dask.apply(lambda x: foo(x, df_first_scattered), axis = 1, meta=('baz', 'int64'))
df_second_dask = df_second_dask.compute()
df_second_dask.head()

実際、このナイーブな dask 実装は、より大きな問題のインスタンスでは、単純な pandas よりも遅いようです。実行パフォーマンスが遅いのは、上記の問題に関連していると思われます。