Python，Dask - 使用来自另一个模块的函数并映射到 Dask Dataframe

Question

我正在尝试使用 map_partitions 在 dask 数据帧上应用一堆函数。当函数在本地定义时有效，例如：

#assume a data frame df1

def upper(x):
    return x.str.upper()

def process(df,info): 
    for mapper,col in info['process']:
        df[col] = df[col].map_partitions(mapper, meta=df[col])
    return df

info = {'process':[(upper, 'column_name')]}
df = process(df1, info)

df.head()

但是当我们将代码拆分成模块时它不起作用..

helper.py

def upper(x):
    return x.str.upper()

def upper_lambda():
    return lambda x: x.str.upper()

main.py

import helper

#assume a data frame df1

def process(df,info): 
    for mapper,col in info['process']:
        df[col] = df[col].map_partitions(mapper, meta=df[col])
    return df

info = {'process':[(getattr(helper,'upper'), 'column_name')]}
#Tried with the lambda too.. dosent seem to work 
#info = {'process':[(helper.upper(), 'column_name')]}

df = process(df1, info)

df.head()

它只是返回 KilledWorker:("('assign-read-parquet-head-1-5-assign-77bd7b855e5e8eec82312c65361fc7c5', 0)",

Answer 1

Dask 当然支持使用来自另一个模块的函数。但是，这些模块应该存在于您正在使用的所有机器上。

对于像 helper.py 文件这样的小文件，您可能需要查看 Client.upload_file 以帮助您移动它。

Python，Dask - 使用来自另一个模块的函数并映射到 Dask Dataframe

Python, Dask - Using functions from another module and mapping to Dask Dataframe

python

python-3.x

pandas

dask

dask-dataframe