InvalidIndexError dask数据帧

Question

我正在学习dask，并在这里和那里得到这个错误：

InvalidIndexError: Reindexing only valid with uniquely valued Index objects

有一个预处理的dask df，我用它做了很多操作，但是一些简单的操作会引发这个错误。

我尝试重新设置索引，但似乎没有帮助。

问题：对于可能出现的问题，是否有高lvl的答案？我能在哪里读到它？为什么经常发生这种情况？谷歌搜索没有帮助。

例如，这里有一个奇怪的例子：我在每次操作后测试df统计数据，以可能看到任何可疑的东西。

df = load_data()
df.shape[0].compute(), df.npartitions
#ok

df = prepare_target(df)
df.shape[0].compute(), df.npartitions
#ok

x_train, x_test, y_train, y_test = dask_tts(df.drop('target', 1), df['target'], random_state=1)
#ok

x_train['target'] = y_train
x_test['target'] = y_test
#ok

x_train.shape[0].compute(), x_train.npartitions
x_test.shape[0].compute(), x_test.npartitions
#ok

x_train.index.nunique().compute()
x_test.index.nunique().compute()
#ok

train, smooth_dict = smoothed_likelyhood(x_train) # returns df and dict
train.shape[0].compute()
#ok

test, _ = smoothed_likelyhood(x_test)
test.shape[0].compute()
#ok

train.index.nunique().compute()
#ok
test.index.nunique().compute() # after this line - error
# InvalidIndexError: Reindexing only valid with uniquely valued Index objects

要指出的是，只有测试才会抛出错误

在这里，我试图重现它，但它的工作原理是：

import numpy as np
import pandas as pd
import dask.dataframe as dd
from dask_ml.model_selection import train_test_split

def smoothed_likelyhood(df, alpha=1): # works with dask df
    global_mean = df['target'].mean()
    smooth_dict = {}

    final_df = df.copy()

    for c in [c for c in df.columns if c!='target']:
        n_rows = df[c].value_counts()
        all_means = df.groupby(by=c)['target'].mean()
        temp_result = ((all_means * n_rows + global_mean + alpha) / (n_rows + alpha))    
        final_df[c] = df[c].map(temp_result)
        smooth_dict[c] = temp_result.compute().to_dict()
        
    return final_df, smooth_dict

# TOY EXAMPLE
test = pd.DataFrame({'a':['mos', 'mos', 'london', 'dubai', 'ny', 'mos', 'london', 'dubai', 'shanghai', 'dubai', 'mos', 'london', 'dubai', 'dubai'],
                     'b':['big', 'big', 'big', 'med', 'med', 'med', 'small', 'small', 'small', 'small', 'big', 'med', 'small', 'med'],
                     'target':[1,0,0,1,0,1,1,0,1,1,1,0,0,0]}) 

df = dd.from_pandas(test, npartitions=2)
# -----------------------------------------------

print(f'npartitions: {df.npartitions}')
x_train, x_test, y_train, y_test = train_test_split(df.drop('target', 1), df['target'], test_size=0.3, shuffle=True)

x_train['target'] = y_train
x_test['target'] = y_test

print(x_train.shape[0].compute(), x_train.index.nunique().compute())
print(x_test.shape[0].compute(), x_test.index.nunique().compute())

train, smooth_dict = smoothed_likelyhood(x_train)
test, _ = smoothed_likelyhood(x_test)

print(train.shape[0].compute(), train.index.nunique().compute())
print(test.shape[0].compute(), test.index.nunique().compute())

# train.compute()

print(train['target'].mean().compute())
print(test['target'].mean().compute())

这部分工作正常，但当我尝试对真实数据执行相同操作时：

%%time
df = load_data(stage='prep_2', frac=config.FRAC, verbose=False) # loading many parquet files
df = prepare_target(df) # some small changes to df
# -----------------------------------------------

print(f'npartitions: {df.npartitions}')
x_train, x_test, y_train, y_test = train_test_split(df.drop('target', 1), df['target'], random_state=1)

x_train['target'] = y_train
x_test['target'] = y_test

print(x_train.shape[0].compute(), x_train.index.nunique().compute())
print(x_test.shape[0].compute(), x_test.index.nunique().compute())

train, smooth_dict = smoothed_likelyhood(x_train)
test, _ = smoothed_likelyhood(x_test)

print(x_train.shape[0].compute(), x_train.index.nunique().compute())
print(x_test.shape[0].compute(), x_test.index.nunique().compute())

这些是打印的结果：

npartitions: 10
1476758 164300
164018 106750
1476758 164300
164018 106750

这2项中的任何一项都会抛出上面提到的索引错误：

train['target'].mean().compute()
test['target'].mean().compute()

如果你有什么想法，我可能会进一步调查，除非我知道去哪里找。

谢谢

Answer 1

问题出现在平滑函数中：

final_df[c] = df[c].map(temp_result)
smooth_dict[c] = temp_result.compute().to_dict()

和DASK系列有关temp_result

通常，对于pandas，如果我想用df中的其他值替换一些值（映射它们），我会这样做：

dict = {'a':{'a1':1, 'a2':2...}, 'b':{'b1':1, 'b2':2...}}
for c in df.columns: # df.columns = ['a', 'b', ...]
    df[c] = df[c].map(dict[c].get) # swap it inplace

它超快

但在这里，在dask，我无法将temp_结果转换为dict（仅在计算之后）。我尝试了一些方法，令人惊讶的是，简单的df.map（series）成功了。正如在映射系列中一样，不是dict.Ok，但这引入了带有无效索引的超级狡猾的问题。

解决方案是使用计算的dict（平滑dict）作为熊猫，而不是系列（temp_结果），并用它进行映射。我认为在dask中过早地计算某些东西是次优的，但我需要结果，而且在这里，当我调用它时，它只计算一次。

以下是一个工作函数：

def smoothed_likelyhood(df, alpha=1):
    """
    Discription:
        preprocess based on mean values of positive events in each category
    Args:
        df: [df] - df to encode
        alpha: [int/float] - regularization param. We can find it with CV
    Returns:
        encoded df, dict to encode user during prediction
    """
    
    global_mean = df['target'].mean()
    smooth_dict = {}

    for c in [c for c in df.columns if c!='target']:
        n_rows = df[c].value_counts()
        all_means = df.groupby(by=c)['target'].mean()
        temp_result = ((all_means * n_rows + global_mean + alpha) / (n_rows + alpha))
        smooth_dict[c] = temp_result.compute().to_dict()
        df[c] = df[c].map(smooth_dict[c])
        
    return df, smooth_dict

尽管如此，它需要25分钟来咀嚼800k*90的数据帧，这只是我数据的5%。如果有人能建议如何在dask中加速这个功能，那将不胜感激。