我們有一個機器學習分類器模型，使用 pandas 數據框和標準 sklearn 管道（StandardScaler、RandomForestClassifier、GridSearchCV 等）進行訓練。我們正在開發 Databricks，并希望使用 Spark 提供的并行計算功能將該管道擴展到大型數據集。將 sklearn 管道轉換為并行計算的最快方法是什么？（我們可以根據需要輕松地在 pandas 和 Spark DF 之間切換。）就上下文而言，我們的選擇似乎是：使用MLLib重寫管道（耗時）使用 sklearn-spark 橋接庫對于選項 2，Spark-Sklearn 似乎已被棄用，但 Databricks建議我們使用 joblibspark。然而，這在 Databricks 上引發了一個例外：from sklearn import svm, datasetsfrom sklearn.model_selection import GridSearchCVfrom joblibspark import register_sparkfrom sklearn.utils import parallel_backendregister_spark() # register spark backendiris = 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
2 回答
MM們
TA貢獻1886條經驗 獲得超2個贊
必要的要求是：
Python 3.6+
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.? ? ? ? ]
12345678_0001
TA貢獻1802條經驗 獲得超5個贊
這個答案對于標準 Spark / Databricks 設置來說應該是正確的，因此考慮到我的問題的措辭/對其他讀者的潛在有用性，我已經接受了它
發現我們案例中的問題后，貢獻一個單獨的“答案”：Databricks 支持人員建議我們案例中的問題是由于我們使用特殊類型的集群（在 AWS 上啟用了憑證直通的高并發性）。grid.fit() 沒有被列入此類集群的白名單，Databricks 建議他們需要向工程團隊提出將其列入白名單。
添加回答
舉報
0/150
提交
取消