如何将sklearn管道转换为pyspark管道？

编程问答 2022-08-13

问题描述

我们有一个机器学习分类器模型，该模型已通过pandas数据框和标准sklearn管道（StandardScaler，RandomForestClassifier，gridsearchcv等）进行了训练。我们正在研究Databricks，并希望使用并行计算Spark提议将此管道扩展到大型数据集。

将sklearn管道转换为可并行计算的东西的最快方法是什么？ （我们可以根据需要轻松地在熊猫和Spark DF之间切换。）

对于上下文，我们的选择似乎是：

  1. 使用MLLib（耗时）重写管道
  2. 使用sklearn-spark桥接库

在选项2上，Spark-Sklearn似乎是deprecated，但我们使用joblibspark的是Databricks recommends。但是，这在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的说明（here和here），必要的要求是：

  • Python 3.6 +
  • pyspark>=2.4
  • scikit-learn>=0.21
  • joblib>=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)

here中的替代示例也是如此：

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.        ]
,

感谢desertnaut的回答-这个答案对于标准的Spark / Databricks设置应该是正确的，因此，鉴于我的问题的措辞/对其他读者的潜在用处，

提供了一个单独的“答案”，从而发现了我们所遇到的问题：Databricks支持人员认为，我们所遇到的问题是由于我们使用了一种特殊类型的集群（在AWS上启用了凭证传递的高并发性）所致。对于这种类型的集群，没有将grid.fit（）列入白名单，Databricks建议他们需要与工程团队一起将其列入白名单。

apache-spark-mldatabrickspysparkpysparkpythonscikit-learn