java.lang.StackOverflowError выбрасывает в spark-submit, но не работает в IDE - PullRequest
Новая
       35

java.lang.StackOverflowError выбрасывает в spark-submit, но не работает в IDE

0 голосов
/ 07 сентября 2018

Я разработал приложение Spark 2.2 для совместной фильтрации. Он прекрасно работает в IntelliJ для запуска или отладки. Я могу войти в Spark Web UI, чтобы проверить процесс. Но когда я попытался развернуть его в EMR и локально протестировать spark-submit, программа не работает должным образом.

Часть команды отправки искры: 

spark-submit -v --master local[*] --deploy-mode client --executor-memory 4G --num-executors 10 --conf spark.executor.extraJavaOptions="-Xss200M " --conf spark.executor.memory="500M" 

def finalStep(sc: SparkContext): Unit = {
        val sameModel = MatrixFactorizationModel.load(sc, "CollaborativeFilter")
        val globalInterestStats = mutable.Map[
            Int, (DescriptiveStatistics, mutable.MutableList[Rating])
        ]()

        val taxonsForUsers = sameModel.recommendProductsForUsers(200)

        taxonsForUsers
            .collect()
            .flatMap(userToInterestArr => {
                userToInterestArr._2.map(rating => {
                    if (globalInterestStats.get(rating.product).isEmpty) {
                        globalInterestStats(rating.product) = (
                            new DescriptiveStatistics(),
                            mutable.MutableList[Rating]()
                        )
                    }

                    globalInterestStats(rating.product)._1.addValue(rating.rating)

                    (rating, userToInterestArr._2)
                })
            })
            .foreach(ratingToUserInterestArr => {
                val rating = ratingToUserInterestArr._1

                if (globalInterestStats.get(rating.product).isDefined) {
                    val interestStats = globalInterestStats(rating.product)
                    val userInterests = ratingToUserInterestArr._2

                    if (rating.rating >= interestStats._1.getPercentile(75)) {
                        userInterests.foreach(each => interestStats._2 += each)
                    }
                }
            })

        println(globalInterestStats.toSeq.length) // ~300

        val globalInterestRDD = sc.parallelize(globalInterestStats.toSeq, 100)// No. of partition does not matter
        val nGlobalInterests = globalInterestStats.map(each => each._2._2.length).sum

// It was not working in spark-submit but I managed to convert this part of code to simplify code before creating the RDD
        val taxonIDFMap = sc.parallelize(
                globalInterestStats
                    .toSeq
                    .flatMap(each => {
                        each._2._2
                            .foldLeft(mutable.Map[Int, Double]())(op = (accu, value) => {
                                if (accu.get(value.product).isEmpty) {
                                    accu(value.product) = 1
                                } else {
                                    accu(value.product) += 1
                                }

                                accu
                            })
                            .toList
                }), 100)
            .reduceByKey((accu, value) => accu + value)
            .map(each => {
                val a: Double = Math.log10(nGlobalInterests / (1 + each._2)) / Math.log10(2)

                (
                    each._1,
                    a
                )
            })
            .collect()
            .toMap

// Yet I have a way more complicated task need to operate on globalInterestRDD which I cannot simplify the size for Spark to handle
        val result = globalInterestRDD
            .count()

        sc.stop()

        println(result)
    }

Exception in thread "dispatcher-event-loop-1" java.lang.StackOverflowError
    at java.io.ObjectOutputStream$ReplaceTable.lookup(ObjectOutputStream.java:2399)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1113)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    ...

Я полагаю, это тесно связано с: http://asyncified.io/2016/12/10/mutablelist-and-the-short-path-to-a-stackoverflowerror/

Но я все еще пытаюсь понять и исправить свой код

1 Ответ

0 голосов
/ 25 марта 2019

Проблема 

val globalInterestStats = mutable.Map[
    Int, (DescriptiveStatistics, mutable.MutableList[Rating])
]()

Должно быть 

val globalInterestStats = mutable.Map[
    Int, (DescriptiveStatistics, mutable.ArrayBuffer[Rating])
]()

Хотя все еще не имеет смысла, почему приложение spark работает в IDE, а не в spark-submit

...