Graphx на Spark создает исключение StackOverflowError в цикле

Question

Я использую Graphx в Scala, чтобы сделать некоторые вычисления.Мне нужно удалить некоторые вершины из графа и сделать несколько циклов.Но код всегда выбрасывает java.lang.StackOverflowError

        var rawG = GraphLoader.edgeListFile(sc, inputFilePath,
            edgeStorageLevel = StorageLevel.MEMORY_AND_DISK, vertexStorageLevel = StorageLevel.MEMORY_AND_DISK)
        while (rawG.vertices.count() > 0) {
            val stepDegree = rawG.degrees.map(v => v._2).min()
            val g = rawG.joinVertices[Int](rawG.degrees)((_, _, newDeg) => newDeg)
            rawG = g.subgraph(vpred = (vid, nodeDegree) => {
                nodeDegree > stepDegree
            })
        }

Исходный граф имеет 100 миллионов вершин.цикл this каждый раз, когда график будет все меньше и меньше, но ошибка StackOverflowError всегда появляется после 100+ циклов.Я полагаю, используя g.subgraph для создания нового подграфа, но память объекта старого графа не освобождается немедленно, поэтому выведите исключение?

Я пытаюсь установить уровень хранения графа как StorageLevel.MEMORY_AND_DISK, ноон выдает ошибки: " Невозможно изменить уровень хранения СДР после того, как ему уже был назначен уровень "

scala> g.checkpoint

scala> g
res15: org.apache.spark.graphx.Graph[Int,Int] = org.apache.spark.graphx.impl.GraphImpl@79863795

scala> g.persist(StorageLevel.MEMORY_AND_DISK)
java.lang.UnsupportedOperationException: Cannot change storage level of an RDD after it was already assigned a level
  at org.apache.spark.rdd.RDD.persist(RDD.scala:170)
  at org.apache.spark.rdd.RDD.persist(RDD.scala:195)
  at org.apache.spark.graphx.impl.VertexRDDImpl.persist(VertexRDDImpl.scala:57)
  at org.apache.spark.graphx.impl.VertexRDDImpl.persist(VertexRDDImpl.scala:27)
  at org.apache.spark.graphx.impl.GraphImpl.persist(GraphImpl.scala:54)
  ... 53 elided

Я прочитал код GraphX, вызов метода Graph.persist RDD.persist (newLevel: StorageLevel) наконец, RDD.isLocallyCheckpointed всегда возвращает false, поэтому RDD.persist () не может изменить уровень хранилища.