无法在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对象和实例是可以选择的