如何将 sklearn 管道转换为 pyspark 管道?
Posted
技术标签:
【中文标题】如何将 sklearn 管道转换为 pyspark 管道?【英文标题】:How to convert a sklearn pipeline into a pyspark pipeline? 【发布时间】:2020-12-20 13:08:55 【问题描述】:我们有一个机器学习分类器模型,我们使用 pandas 数据框和标准 sklearn 管道(StandardScaler、RandomForestClassifier、GridSearchCV 等)进行了训练。我们正在开发 Databricks,并希望使用 Spark 提供的并行计算将此管道扩展到大型数据集。
将我们的 sklearn 管道转换为并行计算的最快方法是什么? (我们可以根据需要在 pandas 和 spark DFs 之间轻松切换。)
就上下文而言,我们的选择似乎是:
-
使用 MLLib 重写管道(耗时)
使用 sklearn-spark 桥接库
在选项 2 上,Spark-Sklearn 似乎是 deprecated,但 Databricks 而不是 recommends,我们使用 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
【问题讨论】:
【参考方案1】:根据 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'
通过上述设置,您的代码 sn-p 运行顺利,并且确实生成了一个分类器 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. ]
【讨论】:
【参考方案2】:感谢desertnaut 的回复 - 这个答案对于标准 Spark / Databricks 设置应该是正确的,因此接受了它,因为我的问题的措辞/对其他读者的潜在用途
在发现我们案例中的问题后提供一个单独的“答案”:Databricks 支持人员表示,我们案例中的问题是由于我们使用了一种特殊类型的集群(在 AWS 上启用了凭证传递的高并发)。 grid.fit() 没有被列入此类集群的白名单,Databricks 建议他们需要与他们的工程团队一起提出它以将其列入白名单。
【讨论】:
以上是关于如何将 sklearn 管道转换为 pyspark 管道?的主要内容,如果未能解决你的问题,请参考以下文章
mlflow 如何使用自定义转换器保存 sklearn 管道?
将数据帧从 pandas 转换为 pyspark 到 Foundry 的数据类型