使用 lambda 函数添​​加列时的 Dask map_partitions meta

Question

我正在使用Dask 应用一个函数myfunc，它将两个新列new_col_1 和new_col_2 添加到我的Dask 数据框data。该函数使用a1 和a2 两列来计算新列。

ddata[['new_col_1', 'new_col_2']] = ddata.map_partitions(
lambda df: df.apply((lambda row: myfunc(row['a1'], row['a2'])), axis=1, 
                    result_type="expand")).compute()  

这会产生以下错误：

ValueError: Metadata inference failed in `lambda`.

You have supplied a custom function and Dask is unable to  determine the type of output that that function returns. 

To resolve this please provide a meta= keyword.

如何为这种情况提供meta 关键字？

Answer 1

meta可以通过kwarg提供给.map_partitions:

some_result = dask_df.map_partitions(some_func, meta=expected_df)

expected_df 可以手动指定，或者您可以在一小部分数据上显式计算它（在这种情况下，它将是一个pandas 数据帧）。

docs有更多详情。

Answer 2

关于使用meta，Sultan 的回答非常完美。 :)

你也可以避免在这里使用map_partitions，因为Dask实现了apply，它在内部调用了map_partitions：

import json
import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({'x': range(1,5),
                   'y': range(6,10),
                  }).astype('str')

ddf = dd.from_pandas(df, npartitions=2)

def myfunc(x):
    s = "string: " + x[0]
    j = json.dumps({'json': x[1]})
    return [s, j]

ddf[['new_col_1', 'new_col_2']] = ddf.apply(myfunc, axis=1, result_type="expand", meta={0: 'object', 1: 'object'})

ddf.compute()

# Output of ddf.compute():
#
#    x  y  new_col_1      new_col_2
# 0  1  6  string: 1  {"json": "6"}
# 1  2  7  string: 2  {"json": "7"}
# 2  3  8  string: 3  {"json": "8"}
# 3  4  9  string: 4  {"json": "9"}

此外，在您的代码 sn-p 中，调用 .compute() 将创建一个 pandas DataFrame，因此，如果将其分配给 Dask DataFrame (ddata)，您将收到错误消息。我建议分配后在ddata 上致电compute。