apache spark - Spark da un StackOverflowError cuando se entrena usando ALS

apache-spark pyspark

Question

Cuando intentaba entrenar un modelo de aprendizaje automático usando ALS en MLLib de Spark, seguía recibiendo un StackoverflowError. Aquí hay una pequeña muestra del seguimiento de la pila:

Traceback (most recent call last):
  File "/Users/user/Spark/imf.py", line 31, in <module>
    model = ALS.train(rdd, rank, numIterations)
  File "/usr/local/Cellar/apache-spark/1.3.1_1/libexec/python/pyspark/mllib/recommendation.py", line 140, in train
    lambda_, blocks, nonnegative, seed)
  File "/usr/local/Cellar/apache-spark/1.3.1_1/libexec/python/pyspark/mllib/common.py", line 120, in callMLlibFunc
    return callJavaFunc(sc, api, *args)
  File "/usr/local/Cellar/apache-spark/1.3.1_1/libexec/python/pyspark/mllib/common.py", line 113, in callJavaFunc
    return _java2py(sc, func(*args))
  File "/usr/local/Cellar/apache-spark/1.3.1_1/libexec/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
  File "/usr/local/Cellar/apache-spark/1.3.1_1/libexec/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o35.trainALSModel.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 40.0 failed 1 times, most recent failure: Lost task 0.0 in stage 40.0 (TID 35, localhost): java.lang.StackOverflowError
        at java.io.ObjectInputStream$PeekInputStream.peek(ObjectInputStream.java:2296)
        at java.io.ObjectInputStream$BlockDataInputStream.peek(ObjectInputStream.java:2589)

Este error también aparecería al intentar ejecutar .mean () para calcular el error cuadrático medio. Apareció tanto en la versión 1.3.1_1 como en la versión 1.4.1 de Spark. Estaba usando PySpark, y aumentar la memoria disponible no ayudó.

Answer 1

La solución fue agregar puntos de verificación, lo que evita que la recursión utilizada por la base de código cree un desbordamiento. Primero, cree un nuevo directorio para almacenar los puntos de control. Luego, haga que su SparkContext use ese directorio para la comprobación de puntos. Aquí está el ejemplo en Python:

sc.setCheckpointDir(''checkpoint/'')

Es posible que también necesite agregar puntos de verificación al ALS, pero no he podido determinar si eso marca la diferencia. Para agregar un punto de control allí (probablemente no sea necesario), simplemente haga lo siguiente:

ALS.checkpointInterval = 2