我们有一个机器学习分类器模型，该模型已通过pandas数据框和标准sklearn管道（StandardScaler，RandomForestClassifier，GridSearchCV等）进行了训练。我们正在研究Databricks，并希望使用并行计算的火花将这个管道扩展到一个大型数据集。
将sklearn管道转换为并行计算的最快方法是什么？（我们可以根据需要轻松地在熊猫和Spark DF之间切换。）
对于上下文，我们的选择似乎是：
在选项2上，Spark-Sklearn似乎已被弃用，但Databricks 建议我们使用joblibspark。但是，这在Databricks上引发了一个例外：
from sklearn import svm, datasets
from sklearn.model_selection import GridSearchCV
from joblibspark import register_spark
from sklearn.utils import parallel_backend
register_spark() # register spark backend
iris = datasets.load_iris()
parameters = {'kernel':('linear', 'rbf'), 'C':[1, 10]}
svr = svm.SVC(gamma='auto')
clf = GridSearchCV(svr, parameters, cv=5)
with parallel_backend('spark', n_jobs=3):
clf.fit(iris.data, iris.target)
加薪
py4j.security.Py4JSecurityException: Method public int org.apache.spark.SparkContext.maxNumConcurrentTasks() is not whitelisted on class class org.apache.spark.SparkContext
根据Databricks的说明（此处和此处），必要的要求是：
pyspark>=2.4scikit-learn>=0.21joblib>=0.14我无法在运行Python 3.7.5，Spark 3.0.0，scikit-learn 0.22.1和joblib 0.14.1的社区Databricks集群中重现您的问题：
import sys
import sklearn
import joblib
spark.version
# '3.0.0'
sys.version
# '3.7.5 (default, Nov 7 2019, 10:50:52) \n[GCC 8.3.0]'
sklearn.__version__
# '0.22.1'
joblib.__version__
# '0.14.1'
使用以上设置，您的代码段可以平稳运行，并确实产生如下分类器clf：
GridSearchCV(cv=5, error_score=nan,
estimator=SVC(C=1.0, break_ties=False, cache_size=200,
class_weight=None, coef0=0.0,
decision_function_shape='ovr', degree=3,
gamma='auto', kernel='rbf', max_iter=-1,
probability=False, random_state=None, shrinking=True,
tol=0.001, verbose=False),
iid='deprecated', n_jobs=None,
param_grid={'C': [1, 10], 'kernel': ('linear', 'rbf')},
pre_dispatch='2*n_jobs', refit=True, return_train_score=False,
scoring=None, verbose=0)
正如这里的替代示例所示：
from sklearn.utils import parallel_backend
from sklearn.model_selection import cross_val_score
from sklearn import datasets
from sklearn import svm
from joblibspark import register_spark
register_spark() # register spark backend
iris = datasets.load_iris()
clf = svm.SVC(kernel='linear', C=1)
with parallel_backend('spark', n_jobs=3):
scores = cross_val_score(clf, iris.data, iris.target, cv=5)
print(scores)
给予
[0.96666667 1. 0.96666667 0.96666667 1. ]
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句