daskデータフレームの.joinの結果が途中で変わるようですが、daskデータフレームが生成されました。

Question

joinを適用して、.from_delayedメソッドで生成されたデータフレームを破棄しました。私は次の例でこれを実証したいと思います。これは3つの部分で構成されています。

1. from_delayed方法を介しDASKのデータフレームを生成し、compute方法でデータフレームをパンダするために、両方のデータフレームに変換from_pandas
2. 介して生成DASKのデータフレームとそれに参加します。 （1）のように参加しました
3. from_delayedメソッドで生成されたdaskデータフレームをcomputeを使用するパンダに変換します。 Aferwardsはfrom_pandasを使用してそれをdaskに戻します。その後、（1）のように参加します。

次のコードを考えてみましょう：

import dask.dataframe 
import pandas as pd 

# functions for generating a dask dataframe 
def get_pdf(character): 
    '''constructs a pandas dataframe with indexes [character]1, ..., [character]5''' 
    index = [character + str(i) for i in range(5)] 
    return pd.DataFrame({'A':[1,2,3,4,5]}, index = index) 

def get_ddf(): 
    '''constructs dask dataframe out of pandas dataframes via the .from-delayed method with indexes A1, A2, A3, ... F3, F3, F4''' 
    delayed_list = [dask.delayed(get_pdf)(x) for x in 'ABCDEF'] 
    return dask.dataframe.from_delayed(delayed_list) 

#generate dask dataframes, that will be joined 
ddf1 = get_ddf() 
ddf2 = dask.dataframe.from_pandas(pd.DataFrame({'B': [1,2,3]}, index = ['A0', 'B1', 'C3']), npartitions = 2) 

#recreate ddf1 by converting it to a pandas dataframe and afterwards to a dask dataframe 
ddf1_from_pandas = dask.dataframe.from_pandas(ddf1.compute(), npartitions = 3) 

#compute joins 
dask_from_delayed_join = ddf1.join(ddf2, how = 'inner') 
pandas_join = ddf1.compute().join(ddf2.compute(), how = 'inner') 
dask_from_pandas_join = ddf1_from_pandas.join(ddf2, how = 'inner') 

を私が同じになるように、すべての3つの結果（dask_from_delayed_join、pandas_join、dask_from_pandas_join）を期待します。

しかし、最初の結果は、他とは異なる：

print(dask_from_delayed_join.compute())：

Empty DataFrame 
Columns: [A, B] 
Index: [] 

print(pandas_join)：

A B 
A0 1 1 
B1 2 2 
C3 4 3 

print(dask_from_pandas_join.compute())：

A B 
A0 1 1 
B1 2 2 
C3 4 3 

何が起こっているのですか？

Answer 1

実際にdd.mergeに問題がありました。これらはdaskバージョン0.10.2で解決されました

In [10]: print(dask_from_delayed_join.compute()) 
    A B 
A0 1 1 
B1 2 2 
C3 4 3 

In [11]: print(pandas_join) 
    A B 
A0 1 1 
B1 2 2 
C3 4 3 

In [12]: print(dask_from_pandas_join.compute()) 
    A B 
A0 1 1 
B1 2 2 
C3 4 3