следующая ситуация.Цель состоит в том, чтобы создать очень простую модель Collaborative Filtering в pyspark на основе заданных частот транзакций для комбинации клиент-продукт.
Я использую класс ALS и ParamGridBuilder, чтобы найти хорошие настройки для модели CF.Однако, когда я запускаю свой код, обучение терпит неудачу, я получаю ошибку, которую я не понимаю.
Вот мой код:
import pandas as pd
import numpy as np
import databricks.koalas as ks
import pyspark.sql.functions as F
from pyspark.sql.types import IntegerType
from pyspark.sql import SQLContext
sqlContext = SQLContext(spark.sparkContext)
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder
# prepare data in the way that we customer, product and "normed" frequency as kind of rating
freq_data = transactions.groupBy(['customer', 'product']).count().withColumnRenamed('count', 'frequency')
customer_purchase_frequency = transactions.groupBy('customer').count().withColumnRenamed('count', 'purchases')
freq_data = freq_data.join(customer_purchase_frequency, on='customer', how='left')
freq_data = freq_data.withColumn('normed_frequency', F.col('frequency')/F.col('purchases'))
freq_data = freq_data.filter(F .col('product').rlike('[0-9]+'))
freq_data = freq_data.withColumn("customer", freq_data["customer"].cast(IntegerType()))
freq_data = freq_data.withColumn("product", freq_data["product"].cast(IntegerType()))
freq_data = freq_data.drop(*['frequency', 'purchases'])
Вывод данных выглядит следующим образом:
display(freq_data.limit(3))
| normed_frequency | customer | product |
|-----------------------|----------|---------|
| 0.024691358024691357 | 36400 | 68398 |
| 0.011741682974559686 | 652 | 68398 |
| 0.0004658746797111577 | 46944 | 68398 |
shape of the data: (3081037, 3)
пока все хорошо.Теперь код, который определяет модель:
# create test and train set
(training, test) = freq_data.randomSplit([0.8, 0.2])
# create ALS model
als = ALS(userCol='customer', itemCol='product', ratingCol='normed_frequency', coldStartStrategy="drop", nonnegative=True)
# Tune model using ParamGridBuilder
param_grid = ParamGridBuilder()\
.addGrid(als.rank, [12, 15, 16, 19])\
.addGrid(als.maxIter, [10, 15, 16, 19])\
.addGrid(als.regParam, [.17, .18, .19, .30])\
.build()
# Define evaluator as RMSE
evaluator = RegressionEvaluator(metricName='rmse', labelCol='normed_frequency', predictionCol="prediction")
# Build cross validation using TrainValidationSplit
tvs = TrainValidationSplit(estimator=als, estimatorParamMaps=param_grid, evaluator=evaluator)
# Fit ALS model to training data
model = tvs.fit(training)
# Extract best model from the tuning exercise using ParamGridBuilder
best_model = model.bestModel
# Generate predictions and evaluate using RMSE
predictions = best_model.transform(test)
rmse = evaluator.evaluate(predictions)
# Print evaluation metrics and model parameters
print('RMSE: ' + str(rmse))
print('-- BEST PARAMETERS --')
print(' Rank: ', best_model.rank)
print(' MaxIter: ', best_model._java_obj.parent().getMaxIter())
print(' Rank: ', best_model._java_obj.parent().getRegParam())
Когда я запускаю код, он останавливается в этой строке:
model = tvs.fit(training)
и возвращает мне эту ошибку, которую я не понимаю:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 11 in stage 3458.0 failed 4 times, most recent failure: Lost task 11.3 in stage 3458.0 (TID 38005, 10.139.64.7, executor 8): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (int) => int)
Я благодарен за каждый ввод, так как я не знаю, где искать отладку.Никогда ранее не использовал pyspark.ml и просто используйте его из-за объема данных.
Вот полный ответ:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<command-1512831702089341> in <module>
20
21 # Fit ALS model to training data
---> 22 model = tvs.fit(training)
23
24 # Extract best model from the tuning exercise using ParamGridBuilder
/databricks/spark/python/pyspark/ml/base.py in fit(self, dataset, params)
130 return self.copy(params)._fit(dataset)
131 else:
--> 132 return self._fit(dataset)
133 else:
134 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)
594 pool = ThreadPool(processes=min(self.getParallelism(), numModels))
595 metrics = [None] * numModels
--> 596 for j, metric, subModel in pool.imap_unordered(lambda f: f(), tasks):
597 metrics[j] = metric
598 if collectSubModelsParam:
/local_disk0/pythonVirtualEnvDirs/virtualEnv-562c2b1c-c0a6-466d-8f33-f7a15de8c57e/lib/python3.7/multiprocessing/pool.py in next(self, timeout)
746 if success:
747 return value
--> 748 raise value
749
750 __next__ = next # XXX
/local_disk0/pythonVirtualEnvDirs/virtualEnv-562c2b1c-c0a6-466d-8f33-f7a15de8c57e/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 <lambda>(f)
594 pool = ThreadPool(processes=min(self.getParallelism(), numModels))
595 metrics = [None] * numModels
--> 596 for j, metric, subModel in pool.imap_unordered(lambda f: f(), tasks):
597 metrics[j] = metric
598 if collectSubModelsParam:
/databricks/spark/python/pyspark/ml/tuning.py in singleTask()
52
53 def singleTask():
---> 54 index, model = next(modelIter)
55 metric = eva.evaluate(model.transform(validation, epm[index]))
56 return index, metric, model if collectSubModel else None
/databricks/spark/python/pyspark/ml/base.py in __next__(self)
60 raise StopIteration("No models remaining.")
61 self.counter += 1
---> 62 return index, self.fitSingleModel(index)
63
64 def next(self):
/databricks/spark/python/pyspark/ml/base.py in fitSingleModel(index)
104
105 def fitSingleModel(index):
--> 106 return estimator.fit(dataset, paramMaps[index])
107
108 return _FitMultipleIterator(fitSingleModel, len(paramMaps))
/databricks/spark/python/pyspark/ml/base.py in fit(self, dataset, params)
128 elif isinstance(params, dict):
129 if params:
--> 130 return self.copy(params)._fit(dataset)
131 else:
132 return self._fit(dataset)
/databricks/spark/python/pyspark/ml/wrapper.py in _fit(self, dataset)
293
294 def _fit(self, dataset):
--> 295 java_model = self._fit_java(dataset)
296 model = self._create_model(java_model)
297 return self._copyValues(model)
/databricks/spark/python/pyspark/ml/wrapper.py in _fit_java(self, dataset)
290 """
291 self._transfer_params_to_java()
--> 292 return self._java_obj.fit(dataset._jdf)
293
294 def _fit(self, dataset):
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling o461810.fit.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 11 in stage 3458.0 failed 4 times, most recent failure: Lost task 11.3 in stage 3458.0 (TID 38005, 10.139.64.7, executor 8): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (int) => int)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:640)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
at scala.collection.Iterator$JoinIterator.hasNext(Iterator.scala:212)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:139)
at org.apache.spark.scheduler.Task.run(Task.scala:112)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1526)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: ALS only supports values in Integer range for columns customer and product. Value null was not numeric.
at org.apache.spark.ml.recommendation.ALSModelParams$$anonfun$4.apply(ALS.scala:102)
at org.apache.spark.ml.recommendation.ALSModelParams$$anonfun$4.apply(ALS.scala:88)
... 19 more
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2355)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2343)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2342)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2342)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1096)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:1096)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1096)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2574)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2522)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2510)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:893)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2243)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2265)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2284)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2309)
at org.apache.spark.rdd.RDD.count(RDD.scala:1184)
at org.apache.spark.ml.recommendation.ALS$.train(ALS.scala:932)
at org.apache.spark.ml.recommendation.ALS$$anonfun$fit$1.apply(ALS.scala:676)
at org.apache.spark.ml.recommendation.ALS$$anonfun$fit$1.apply(ALS.scala:658)
at org.apache.spark.ml.util.Instrumentation$$anonfun$14.apply(Instrumentation.scala:277)
at scala.util.Try$.apply(Try.scala:192)
at org.apache.spark.ml.util.Instrumentation$.instrumented(Instrumentation.scala:277)
at org.apache.spark.ml.recommendation.ALS.fit(ALS.scala:658)
at sun.reflect.GeneratedMethodAccessor621.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
at py4j.Gateway.invoke(Gateway.java:295)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:251)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (int) => int)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:640)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
at scala.collection.Iterator$JoinIterator.hasNext(Iterator.scala:212)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:139)
at org.apache.spark.scheduler.Task.run(Task.scala:112)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$13.apply(Executor.scala:497)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1526)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:503)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: java.lang.IllegalArgumentException: ALS only supports values in Integer range for columns customer and product. Value null was not numeric.
at org.apache.spark.ml.recommendation.ALSModelParams$$anonfun$4.apply(ALS.scala:102)
at org.apache.spark.ml.recommendation.ALSModelParams$$anonfun$4.apply(ALS.scala:88)
... 19 more