Агрегация Spark Cassandra java.lang.OutOfMemoryError: Java heap space

Question

Я пытаюсь научиться использовать Apache Spark, и у меня возникают проблемы с попыткой суммировать все значения в столбце из Cassandra (используя искру datastax -cassandra-разъем). Все, что я пытаюсь сделать, просто приводит к java.lang.OutOfMemoryError: Java heap space.

Вот код, я подчиняясь искрового мастера:

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) 
    } 
} 

Прямо сейчас у меня есть только один узел искрового уборщица в моей группе, и это, безусловно, возможно, что, учитывая размер таблицы, не все это может сразу вписываться в память. Однако я не думал, что это будет проблемой, поскольку искра должна лениво оценивать команды, а суммирование всех значений в столбце не обязательно должно содержать всю таблицу в памяти сразу.

Я очень новичок в этой теме, поэтому любое разъяснение относительно того, почему это не сработает или поможет, как правильно это сделать, было бы очень признательно.

Благодаря

Answer 1

Возможно, искра строит всю таблицу как один в раздел памяти, так что она может делать операции отображения на нем.

Я думал, что искра должна была разливаться на диск, а не вызывать 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"); 

Так попробуйте установить spark.cassandra.input.split.size_in_mb в вашем конф.

Я предполагаю, что это позволило бы искра суммировать куски стола, а затем высекать эти куски из памяти, когда ему нужно место для новых кусков.

Еще одна вещь, которую вы могли бы изучить, - это указать уровень хранения для таблицы RDD, которая позволила бы ей разлиться на диск. Я думаю, вы могли бы сделать это, добавив «.persist (StorageLevel.MEMORY_AND_DISK)». По умолчанию используется MEMORY_ONLY. См. Дополнительную информацию о уровнях хранения here в разделе «Сохранение RDD».