I'm trying to tune the parameters of an ALS matrix factorization model.
So I'm using pyspark.ml.tuning.CrossValidator to run through a parameter grid and select the best model.
But When tuning with CrossValidator, I always get ERROR..
The error is as shown in the photo below, and referring to the answer in the following stackoverflow article, it seems that an error occurs when inferring users who are not in the train data from the valid data in the ALS model.
Also, referring to the following article, it is assumed that CrossValidator causes an error and stops the main thread when a fitting error occurs when it is running on multiple nodes in parallel, but I don't know how to solve it.
When I apply GridSearch with a for statement directly, there is no error, so I don't know why the error below only occurs in the CrossValidator.
Also, the ALS hyperparameter coldStartStrategy = "drop" seems to prevent errors in data that are unseen in the train data. I wonder if this part applies only to items.
from pyspark.ml.recommendation import ALS
als = ALS(
userCol = "user" # user
, itemCol = "item" # item
, ratingCol = 'read' # interaction log (read : 0 or 1)
, coldStartStrategy = "drop" #
, implicitPrefs = True # implicit feedback
, nonnegative = True # Non-Negative Factorization
)
from pyspark.ml.tuning import TrainValidationSplit, CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.evaluation import RegressionEvaluator
ParamMaps = (ParamGridBuilder()
.addGrid(als.rank, [20, 40, 80, 120])
.addGrid(als.maxIter, [10, 30, 50])
.addGrid(als.regParam, [0.001, 0.01, 0.05, 0.1])
).build()
evaluator = RegressionEvaluator( metricName = "rmse"
, labelCol = "read"
, predictionCol = "prediction")
cv = CrossValidator( estimator = als
, estimatorParamMaps = ParamMaps
, evaluator = evaluator
, parallelism = 5
, numFolds = 5
, seed = 42
)
cvmodel = cv.fit(train)
Here is an Error
MLlib will automatically track trials in MLflow. After your tuning fit() call has completed, view the MLflow UI to see logged runs.
/databricks/spark/python/pyspark/ml/util.py:90: UserWarning: CrossValidator_3986dadc628f fit call failed but some spark jobs may still running for unfinished trials. To address this issue, you should enable pyspark pinned thread mode.
.format(uid))
IllegalArgumentException: requirement failed: Nothing has been added to this summarizer.
---------------------------------------------------------------------------
IllegalArgumentException Traceback (most recent call last)
<command-3029966> in <module>
10 if one of the trial tasks failed, the CrossValidator/TrainValidationSplit fit will raise error and break the main thread, but other backgroud threads running other trial tasks will continue to run, and trial tasks which are pending to run in thread queue will also continue to launch.
11 '''
---> 12 cvmodel = cv.fit(train)
/databricks/spark/python/pyspark/ml/base.py in fit(self, dataset, params)
127 return self.copy(params)._fit(dataset)
128 else:
--> 129 return self._fit(dataset)
130 else:
131 raise ValueError("Params must be either a param map or a list/tuple of param maps, "
/databricks/spark/python/pyspark/ml/tuning.py in _fit(self, dataset)
458 subModels[i][j] = subModel
459
--> 460 _cancel_on_failure(dataset._sc, self.uid, sub_task_failed, calculate_metrics)
461 validation.unpersist()
462 train.unpersist()
/databricks/spark/python/pyspark/ml/util.py in _cancel_on_failure(sc, uid, sub_task_failed, f)
89 "issue, you should enable pyspark pinned thread mode."
90 .format(uid))
---> 91 raise e
92
93 old_job_group = sc.getLocalProperty("spark.jobGroup.id")
/databricks/spark/python/pyspark/ml/util.py in _cancel_on_failure(sc, uid, sub_task_failed, f)
83 if os.environ.get("PYSPARK_PIN_THREAD", "false").lower() != "true":
84 try:
---> 85 return f()
86 except Exception as e:
87 warnings.warn("{} fit call failed but some spark jobs "
/databricks/spark/python/pyspark/ml/tuning.py in calculate_metrics()
452 return task()
453
--> 454 for j, metric, subModel in pool.imap_unordered(run_task, tasks):
455 metrics[j] += (metric / nFolds)
456 metrics_all[i][j] = metric
/usr/lib/python3.7/multiprocessing/pool.py in next(self, timeout)
746 if success:
747 return value
--> 748 raise value
749
750 __next__ = next # XXX
/usr/lib/python3.7/multiprocessing/pool.py in worker(inqueue, outqueue, initializer, initargs, maxtasks, wrap_exception)
119 job, i, func, args, kwds = task
120 try:
--> 121 result = (True, func(*args, **kwds))
122 except Exception as e:
123 if wrap_exception and func is not _helper_reraises_exception:
/databricks/spark/python/pyspark/ml/tuning.py in run_task(task)
450 if sub_task_failed[0]:
451 raise RuntimeError("Terminate this task because one of other task failed.")
--> 452 return task()
453
454 for j, metric, subModel in pool.imap_unordered(run_task, tasks):
/databricks/spark/python/pyspark/ml/tuning.py in singleTask()
57 # `MetaAlgorithmReadWrite.getAllNestedStages`, make it return
58 # all nested stages and evaluators
---> 59 metric = eva.evaluate(model.transform(validation, epm[index]))
60 return index, metric, model if collectSubModel else None
61
/databricks/spark/python/pyspark/ml/evaluation.py in evaluate(self, dataset, params)
70 return self.copy(params)._evaluate(dataset)
71 else:
---> 72 return self._evaluate(dataset)
73 else:
74 raise ValueError("Params must be a param map but got %s." % type(params))
/databricks/spark/python/pyspark/ml/evaluation.py in _evaluate(self, dataset)
100 """
101 self._transfer_params_to_java()
--> 102 return self._java_obj.evaluate(dataset._jdf)
103
104 def isLargerBetter(self):
/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
1303 answer = self.gateway_client.send_command(command)
1304 return_value = get_return_value(
-> 1305 answer, self.gateway_client, self.target_id, self.name)
1306
1307 for temp_arg in temp_args:
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
131 # Hide where the exception came from that shows a non-Pythonic
132 # JVM exception message.
--> 133 raise_from(converted)
134 else:
135 raise
/databricks/spark/python/pyspark/sql/utils.py in raise_from(e)
IllegalArgumentException: requirement failed: Nothing has been added to this summarizer.