Spark Cassandra 聚合 java.lang.OutOfMemoryError：Java 堆空间

Question

我一直在尝试学习如何使用 Apache Spark，但在尝试对 Cassandra 列中的所有值求和时遇到问题（使用 datastax spark-cassandra-connector）。我尝试的所有操作都会导致 java.lang.OutOfMemoryError: Java heap space。

这是我提交给 spark master 的代码：

object Benchmark {
  def main( args: Array[ String ] ) {
    val conf    = new SparkConf()
                  .setAppName( "app" )
                  .set( "spark.cassandra.connection.host", "ec2-blah.compute-1.amazonaws.com" )
                  .set( "spark.cassandra.auth.username", "myusername" )
                  .set( "spark.cassandra.auth.password", "mypassword" )
                  .set( "spark.executor.memory", "4g" )
    val sc      = new SparkContext( conf )
    val tbl     = sc.cassandraTable( "mykeyspace", "mytable" )
    val res     = tbl.map(_.getFloat("sclrdata")).sum()

    println( "sum = " + res )
  }
}

现在我的集群中只有一个 spark 工作节点，而且鉴于表的大小，绝对有可能不是所有的都可以同时放入内存中。但是我不认为这会是一个问题，因为 spark 应该懒惰地评估命令，并且对列中的所有值求和不需要让整个表一次驻留在内存中。

我是这个主题的新手，所以任何关于为什么这不起作用的澄清或关于如何正确执行它的帮助将不胜感激。

谢谢

Answer 1

也许 spark 正在将整个表构建为单个内存分区，以便它可以对其进行映射操作。

我认为 spark 应该溢出到磁盘而不是抛出 OutOfMemoryExceptions，但如果只有一个分区，它可能无法溢出。我看到了类似的问题here，他通过指定这样的分割大小解决了这个问题：

conf = new SparkConf();
        conf.setAppName("Test");
        conf.setMaster("local[4]");
        conf.set("spark.cassandra.connection.host", "192.168.1.15").
        set("spark.executor.memory", "2g").
        set("spark.cassandra.input.split.size_in_mb", "67108864");

所以尝试在你的 conf 中设置 spark.cassandra.input.split.size_in_mb。

我想这将允许 spark 汇总表的块，然后在它需要新块的空间时将这些块从内存中逐出。

您可以研究的另一件事是为表 RDD 指定一个存储级别，以允许它溢出到磁盘。我认为您可以通过添加“.persist(StorageLevel.MEMORY_AND_DISK)”来做到这一点。默认值似乎是 MEMORY_ONLY。请参阅 RDD 持久性部分中有关存储级别 here 的更多信息。