当处理非常大的数据时，设备上没有剩余的空间

Question

val vertex = graph.vertices
val edges = graph.edges.map(v=>(v.srcId, v.dstId)).toDF("key","value")
var FMvertex = vertex.map(v => (v._1, HLLCounter.encode(v._1)))
var encodedVertex = FMvertex.toDF("keyR", "valueR")

var Degvertex = vertex.map(v => (v._1, 0.toLong))
var lastRes = Degvertex
//calculate FM of the next step
breakable {
  for (i <- 1 to MaxIter) {
    var N_pre = FMvertex.map(v => (v._1, HLLCounter.decode(v._2)))
    var adjacency = edges.join(
      encodedVertex,//FMvertex.toDF("keyR", "valueR"),
      $"value" === $"keyR"
    ).rdd.map(r => (r.getAs[VertexId]("key"), r.getAs[Array[Byte]]("valueR"))).reduceByKey((a,b)=>HLLCounter.Union(a,b))
    FMvertex = FMvertex.union(adjacency).reduceByKey((a,b)=>HLLCounter.Union(a,b))

    // update vetex encode
    encodedVertex = FMvertex.toDF("keyR", "valueR")

    var N_curr = FMvertex.map(v => (v._1, HLLCounter.decode(v._2)))
    lastRes = N_curr
    var middleAns = N_curr.union(N_pre).reduceByKey((a,b)=>Math.abs(a-b))//.mapValues(x => x._1 - x._2)
    if (middleAns.values.sum() == 0){
      println(i)
      break
    }
    Degvertex = Degvertex.join(middleAns).mapValues(x => x._1 + i * x._2)//.map(identity)
  }
}
val res = Degvertex.join(lastRes).mapValues(x => x._1.toDouble / x._2.toDouble)
return res

    import net.agkn.hll.HLL;
import com.google.common.hash.*;
import com.google.common.hash.Hashing;

import java.io.Serializable;

public class HLLCounter implements Serializable {
    private static int seed = 1234567;
    private static HashFunction hs = Hashing.murmur3_128(seed);

    private static int log2m = 15;
    private static int regwidth = 5;


    public static byte[] encode(Long id) {
        HLL hll = new HLL(log2m, regwidth);
        Hasher myhash = hs.newHasher();
        hll.addRaw(myhash.putLong(id).hash().asLong());
        return hll.toBytes();
    }

    public static byte[] Union(byte[] byteA, byte[] byteB) {
        HLL hllA = HLL.fromBytes(byteA);
        HLL hllB = HLL.fromBytes(byteB);
        hllA.union(hllB);
        return hllA.toBytes();
    }

    public static long decode(byte[] bytes) {
        HLL hll = HLL.fromBytes(bytes);
        return hll.cardinality();
    }
}

当我在一个大约有一千万个顶点和一亿条边的图上运行时，代码工作得很好。然而，当我在一个有数以亿计的图和数十亿条边的图上运行它时，在集群上运行了几个小时后，它显示

Driver stacktrace:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 91 in stage 29.1 failed 4 times, most recent failure: Lost task 91.3 in stage 29.1 (TID 17065, 9.10.135.216, executor 102): java.io.IOException: : No space left on device
 at java.io.FileOutputStream.writeBytes(Native Method)
 at java.io.FileOutputStream.write(FileOutputStream.java:326)
 at org.apache.spark.storage.TimeTrackingOutputStream.write(TimeTrackingOutputStream.java:58)
 at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)

有人能帮我吗？我刚开始使用spark几天。谢谢你帮忙。

Answer 1

小天，你说“洗牌读和洗牌写大约是1TB。我不需要那些中间值或RDDs”。此声明确认您不熟悉Apache Spark，也可能不熟悉您正在运行的算法。请听我解释。

当三个数相加时，你必须对要相加的前两个数做出选择。例如(A+B)+C或A+(B+C)。一旦做出选择，括号内的数字就会有一个临时的中间值。在没有中间数的情况下，不可能在所有三个数之间继续计算。

RDD是一种空间高效的数据结构。每个“新”RDD表示整个数据集上的一组操作。一些RDDs表示单个操作，如“加五”，而另一些RDDs表示一系列操作，如“加五，然后乘六，再减七”。如果不放弃数学算法的某些部分，就不能放弃RDD。

Apache Spark的核心是一种散集算法。它将数据集分发到多个工作节点，其中该数据集是分发的单个RDD的一部分，以及所需的计算。此时，计算尚未执行。当从RDD的计算形式请求数据时，按需执行计算。

有时，如果不知道其他工人的一些中间值，就不可能在单个工人上完成计算。这种工作人员之间的交叉通信总是发生在头节点之间，头节点将数据分配给各个工作人员，并收集和聚合来自各个工作人员的数据；但是，根据算法的结构，它也可能发生在计算中期（特别是在分组或连接数据片的算法中）。

您有一个算法需要洗牌，以这样一种方式，单个节点不能从所有其他节点收集结果，因为单个节点没有足够的ram来保存从其他节点收集的中间值。

1. 优化RDD中的分区以减少串扰（太小的分区可能需要更多的串扰，因为完全连接的互传输在O(n^2)处增长，太大的分区可能会耗尽计算节点内的ram）。
2. 重新构造算法，使其不需要完全洗牌（有时可以分阶段进行精简，以便处理更多的精简阶段，每个阶段合并的数据更少）。
3. 重新构造算法，使其不需要洗牌（算法可能只是写错了，但不太可能，以不同的方式分解可以避免从节点的角度请求远程数据）。
4. 如果问题出在收集结果上，请重写算法，使结果不返回在头节点的控制台中，而是返回在能够容纳数据的分布式文件系统（如HDFS）中。

如果没有Apache Spark程序的细节，没有对数据集的访问，没有对Spark集群及其日志的访问，很难知道这些常见方法中哪一种对您的好处最大；所以我把它们都列了出来。

祝你好运！