Dask - 如何使用 apply 将 Series 连接到 DataFrame 中？答案

【问题标题】：Dask - How to concatenate Series into a DataFrame with apply?Dask - 如何使用 apply 将 Series 连接到 DataFrame 中？
【发布时间】：2019-03-05 01:42:43
【问题描述】：

如何从应用于 Dask 系列的函数返回多个值？我试图从dask.Series.apply 的每次迭代中返回一个系列，最终结果是dask.DataFrame。

以下代码告诉我元数据错误。然而，全熊猫版本有效。这里有什么问题？

更新：我认为我没有正确指定元/模式。我该如何正确地做到这一点？现在，当我放弃 meta 参数时它可以工作。但是，它会发出警告。我想“正确”使用 dask。

import dask.dataframe as dd
import pandas as pd
import numpy as np
from sklearn import datasets

iris = datasets.load_iris()

def transformMyCol(x):
    #Minimal Example Function
    return(pd.Series(['Tom - ' + str(x),'Deskflip - ' + str(x / 8),'']))

#
## Pandas Version - Works as expected.
#
pandas_df = pd.DataFrame(data= np.c_[iris['data'], iris['target']], columns= iris['feature_names'] + ['target'])
pandas_df.target.apply(transformMyCol,1)

#
## Dask Version (second attempt) - Raises a warning
#
df = dd.from_pandas(pandas_df, npartitions=10)

unpacked = df.target.apply(transformMyCol)
unpacked.head()

#
## Dask Version (first attempt) - Raises an exception 
#
df = dd.from_pandas(pandas_df, npartitions=10)

unpacked_dask_schema = {"name" : str, "action" : str, "comments" : str}

unpacked = df.target.apply(transformMyCol, meta=unpacked_dask_schema)
unpacked.head()

这是我得到的错误：

  File "/anaconda3/lib/python3.7/site-packages/dask/dataframe/core.py", line 3693, in apply_and_enforce
    raise ValueError("The columns in the computed data do not match"
ValueError: The columns in the computed data do not match the columns in the provided metadata

我也验证了以下内容，但它也不起作用。

meta_df = pd.DataFrame(dtype='str',columns=list(unpacked_dask_schema.keys()))


unpacked = df.FILEDATA.apply(transformMyCol, meta=meta_df)
unpacked.head()

同样的错误：

  File "/anaconda3/lib/python3.7/site-packages/dask/dataframe/core.py", line 3693, in apply_and_enforce
    raise ValueError("The columns in the computed data do not match"
ValueError: The columns in the computed data do not match the columns in the provided metadata

【问题讨论】：

您好，请提供mcve。特别是 df 的小样本会很棒。
@user32185 完成。
@user32185 澄清一下：第一段代码现在是引发所描述异常的 MCVE。
我不清楚这个 apply 应该做什么。你有一个使用熊猫的例子吗？或者您可以发布您想要获得的输出吗？
@user32185 完成。我还让transformMyCol() 变得不那么简单了。

标签： python pandas dataframe dask dask-distributed

【解决方案1】：

你是对的，问题是你没有正确指定元数据；更具体地说，正如错误消息所说，元数据列 ("name", "action", "comments") 与计算数据中的列 (0, 1, 2) 不匹配。您应该：

将元数据列更改为 0、1、2：

   unpacked_dask_schema = dict.fromkeys(range(3), str)
   df.target.apply(transformMyCol, meta=unpacked_dask_schema)

或

更改 transformMyCol 以使用命名列：


    def transformMyCol(x):
        return pd.Series({
            'name': 'Tom - ' + str(x), 
            'action': 'Deskflip - ' + str(x / 8), 
            'comments': '',
        }))

【讨论】：