我一直在尝试编写自定义类，用于Preprocessing跟随Feature selection和Machine Learning算法。
我(preprocessing only)使用@delayed. 但是当我从教程中读到相同的内容时，可以使用Client. 它引起了两个问题。
作为脚本运行。不像 Jupyter 笔记本
第一个问题：
# Haven't run any scheduler or worker manually
client = Client() # Nothing passed as an argument
# Local Cluster is not working;
Error:...
if __name__=='__main__':
freeze_support()
...
我在 Jupyter Notebook 中尝试了同样的方法，没有在不同的终端中运行任何调度程序或工作程序。有效！！
现在，我用 1 个调度程序和 2 个工作人员触发了 3 个终端，并将其更改为Client('IP')在脚本中。错误已解决，此行为的任何原因。
第二个问题：
问题标题中提到的错误。将client = Client('IP')作为参数传递给构造函数并将使用self.client.submit的东西传递给集群。但失败并显示错误消息
错误：没有模块名称“diya_info”
这是代码：
主文件
import dask.dataframe as dd
from diya_info import Diya_Info
import time
# from dask import delayed
from dask.distributed import Client
df = dd.read_csv(
'/Users/asifali/workspace/playground/flask/yellow_tripdata_2015- 01.csv')
# df = delayed(df.fillna(0.3))
# df = df.compute()
client = Client('192.168.0.129:8786')
X = df.drop('payment_type', axis=1).copy()
y = df['payment_type']
Instance = Diya_Info(X, y, client)
s = time.ctime(int(time.time()))
print(s)
Instance = Instance.fit(X, y)
e = time.ctime(int(time.time()))
print(e)
# print((e-s) % 60, ' secs')
迪亚信息.py
from sklearn.base import TransformerMixin, BaseEstimator
from dask.multiprocessing import get
from dask import delayed, compute
class Diya_Info(BaseEstimator, TransformerMixin):
def __init__(self, X, y, client):
assert X is not None, 'X can\'t be None'
assert type(X).__name__ == 'DataFrame', 'X not of type DataFrame'
assert y is not None, 'y can\'t be None'
assert type(y).__name__ == 'Series', 'y not of type Series'
self.client = client
def fit(self, X, y):
self.X = X
self.y = y
# X_status = self.has_null(self.X)
# y_status = self.has_null(self.y)
# X_len = self.get_len(self.X)
# y_len = self.get_len(self.y)
X_status = self.client.submit(self.has_null, self.X)
y_status = self.client.submit(self.has_null, self.y)
X_len = self.client.submit(self.get_len, self.X)
y_len = self.client.submit(self.get_len, self.y)
# X_null, y_null, X_length, y_length
X_null, y_null, X_length, y_length = self.client.gather(
[X_status, y_status, X_len, y_len])
assert X_null == False, 'X contains some columns with null/NaN values'
assert y_null == False, 'y contains some columns with null/NaN values'
assert X_length == y_length, 'Shape mismatch, X and y are of different length'
return self
def transform(self, X):
return X
@staticmethod
# @delayed
def has_null(df):
return df.isnull().values.any()
@staticmethod
# @delayed
def get_len(df):
return len(df)
这是完整的堆栈跟踪：
Sat Aug 11 13:29:08 2018
distributed.utils - ERROR - No module named 'diya_info'
Traceback (most recent call last):
File "/anaconda3/lib/python3.6/site-packages/distributed/utils.py", line 238, in f
result[0] = yield make_coro()
File "/anaconda3/lib/python3.6/site-packages/tornado/gen.py", line 1055, in run
value = future.result()
File "/anaconda3/lib/python3.6/site-packages/tornado/concurrent.py", line 238, in result
raise_exc_info(self._exc_info)
File "<string>", line 4, in raise_exc_info
File "/anaconda3/lib/python3.6/site-packages/tornado/gen.py", line 1063, in run
yielded = self.gen.throw(*exc_info)
File "/anaconda3/lib/python3.6/site-packages/distributed/client.py", line 1315, in _gather
traceback)
File "/anaconda3/lib/python3.6/site-packages/six.py", line 692, in reraise
raise value.with_traceback(tb)
File "/anaconda3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 59, in loads
return pickle.loads(x)
ModuleNotFoundError: No module named 'diya_info'
Traceback (most recent call last):
File "notebook/main.py", line 24, in <module>
Instance = Instance.fit(X, y)
File "/Users/asifali/workspace/pythonProjects/ML-engine-DataX/pre-processing/notebook/diya_info.py", line 28, in fit
X_status, y_status, X_len, y_len)
File "/anaconda3/lib/python3.6/site-packages/distributed/client.py", line 2170, in compute
result = self.gather(futures)
File "/anaconda3/lib/python3.6/site-packages/distributed/client.py", line 1437, in gather
asynchronous=asynchronous)
File "/anaconda3/lib/python3.6/site-packages/distributed/client.py", line 592, in sync
return sync(self.loop, func, *args, **kwargs)
File "/anaconda3/lib/python3.6/site-packages/distributed/utils.py", line 254, in sync
six.reraise(*error[0])
File "/anaconda3/lib/python3.6/site-packages/six.py", line 693, in reraise
raise value
File "/anaconda3/lib/python3.6/site-packages/distributed/utils.py", line 238, in f
result[0] = yield make_coro()
File "/anaconda3/lib/python3.6/site-packages/tornado/gen.py", line 1055, in run
value = future.result()
File "/anaconda3/lib/python3.6/site-packages/tornado/concurrent.py", line 238, in result
raise_exc_info(self._exc_info)
File "<string>", line 4, in raise_exc_info
File "/anaconda3/lib/python3.6/site-packages/tornado/gen.py", line 1063, in run
yielded = self.gen.throw(*exc_info)
File "/anaconda3/lib/python3.6/site-packages/distributed/client.py", line 1315, in _gather
traceback)
File "/anaconda3/lib/python3.6/site-packages/six.py", line 692, in reraise
raise value.with_traceback(tb)
File "/anaconda3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 59, in loads
return pickle.loads(x)
ModuleNotFoundError: No module named 'diya_info'
如果我取消注释@delayed和更多评论，它会起作用。client但是如何通过传入作为参数来使其工作。想法是对我要编写的所有库使用相同的客户端。
更新 1：
我second problem通过删除@staticmethod装饰器并将函数放置在fit closure. 但是有什么问题@staticmethod，这些装饰器是为非自我相关的东西设计的，对吧？
这是diya_info.py：
...
def fit(self, X, y):
self.X = X
self.y = y
# function removed from @staticmethod
def has_null(df): return df.isnull().values.any()
# function removed from @staticmethod
def get_len(df): return len(df)
X_status = self.client.submit(has_null, self.X)
y_status = self.client.submit(has_null, self.y)
...
有没有办法做到这一点@staticmethod。我对解决这个问题的方式感觉不好。仍然没有任何线索Problem 1