无法在dask worker中加载模块

我有以下项目树

.
└── src
    └── dask_test
        ├── helpers
        │   ├── commandline.py
        │   ├── data
        │   │   ├── dataset0.json
        │   │   ├── dataset1000.json
        │   │   ├── dataset300.json
        │   │   ├── dataset5000.json
        │   │   ├── dataset500.json
        │   │   ├── events_to_be_used_final_without_google.nl.json
        │   │   ├── http-malware_modified.log
        │   │   └── public_suffix_list.json
        │   ├── datetime.py
        │   ├── datetime.pyc
        │   ├── __init__.py
        │   ├── __init__.pyc
        │   ├── math.py
        │   ├── math.pyc
        │   ├── pipeline.py
        │   ├── queues.py
        │   ├── search.py
        │   ├── services.py
        │   ├── sklearn.py
        │   ├── splunk_format.py
        │   ├── splunk.py
        │   └── sqlalchemy.py
        ├── __init__.py
        ├── __init__.pyc
        ├── main.py
        └── riskanalysis
            ├── iaccess
            │   ├── __init__.py
            │   └── metrics
            │       ├── base.py
            │       ├── __init__.py
            │       └── profile
            │           └── __init__.py
            ├── __init__.py
            └── metrics
                └── __init__.py

main.py 的开头,我从`dask_test.helpers.datetime'导入一个对象

from dask_test.helpers.datetime import Timewindow

在我的主要使用 . 在我的主文件中,我定义了一些函数并将它们应用于这样的dask Dataframe

dataframe = transformation1(dataframe)
dataframe = transformation2(dataframe)
dataframe = transformation3(dataframe)
dataframe = transformation4(dataframe)

转换函数使用dask.dataframe并使用apply它们添加一个新列,如下所示:

def transformation1(dataframe):
    dataframe['new_column'] = dataframe.apply(make_sequence)
    return dataframe

尝试使用dask分布式计算和LocalCluster正常工作:

from distributed import Client, LocalCluster
cluster = LocalCluster(n_workers=4, threads_per_worker=1, processes=True)
client = Client(cluster)
client.persist(dataframe)

但是当打开dask-scheduler和dask-workers时,我收到以下消息

return pickle.loads(x)ImportError:没有名为dask_test.helpers.datetime的模块

  • 拳头问题不是LocalCluster使用酸洗?

  • 所有模块都需要包含pickleable对象才能正确分配dask?

编辑:

导入datetime模块和cloudpickle似乎datetime是可选择的

from dask_test.helpers import datetime
import cloudpickle

cloudpickle.dumps(datetime)  # this works
datetime_module = cloudpickle.loads(cloudpickle.dumps(datetime)) # this also works

编辑:经过一些调查后,我在日志文件中看到了这一点:

distributed.protocol.pickle - INFO - Failed to deserialize �cpandas.core.frame

数据帧

在我的主文件中,我首先创建pandas Dataframe,然后使用 from_pandas 方法将其转换为dask DataFrame .

编辑3:我发现问题是什么,但我不明白为什么 . 在我的datetime模块中,我定义了一个名为TimeWindow的对象来处理时间段 . 我的datajson文件有一个时间戳形式的字段

timestamp_since-timestamp_until

我在数据框上应用一个函数来更改添加一个列,将上面的内容转换为时间窗口对象,如下所示:

def convert_to_time_window(item):
    since = my_datetime.utcfromtimestamp_tzaware(float(item.split('-')[0]))
    until = my_datetime.utcfromtimestamp_tzaware(float(item.split('-')[1]))

    return my_datetime.TimeWindow(tm_since=since, tm_until=until)

在Dataframe上(这是一个pandas数据帧 . 我在创建一个dask数据帧之前这样做)

dataframe['tw'] = dataframe['time_bucket'].apply(convert_to_time_window)

没有它, Worker 工作正常 . 但TimeWindow对象和实例是可以选择的

回答(1)

3 years ago

听起来您的dask-worker进程无法像客户端进程那样访问 dask_test.helpers.datetime 模块 . 从您描述项目的方式来看,您似乎依赖于从与文件相同的位置运行Python进程 . 您可以执行以下任一操作:

  • 将软件安装为正确的Python模块(有关更多信息,请参阅Python文档)

  • 从与客户端进程相同的目录运行dask-worker进程

distributed.protocol.pickle - INFO - 无法反序列化 cpandas.core.frame

这个错误听起来像你的dask-worker进程无法访问Pandas . 通常,您的dask-worker进程需要在任何地方都拥有相同的软件环境 .

验证统一的当前工作目录

要验证所有工作人员是否具有相同的当前工作目录,请尝试在所有工作人员上运行 os.getcwd

In [6]: client.run(os.getcwd)
Out[6]: 
{'tcp://127.0.0.1:34115': '/home/foo',
 'tcp://127.0.0.1:39449': '/home/foo',
 'tcp://127.0.0.1:40322': '/home/foo',
 'tcp://127.0.0.1:41050': '/home/foo'}

您可以将此与在python进程中本地调用 os.getcwd() 进行比较 .