Increasing Spark_executors_instances to 4 worked. SPARK_EXECUTOR_INSTANCES="4" #Number of workers to start (Default: 2)
Regards, Vinti On Wed, Mar 2, 2016 at 4:28 AM, Vinti Maheshwari <vinti.u...@gmail.com> wrote: > Thanks much Saisai. Got it. > So i think increasing worker executor memory might work. Trying that. > > Regards, > ~Vinti > > On Wed, Mar 2, 2016 at 4:21 AM, Saisai Shao <sai.sai.s...@gmail.com> > wrote: > >> You don't have to specify the storage level for direct Kafka API, since >> it doesn't require to store the input data ahead of time. Only >> receiver-based approach could specify the storage level. >> >> Thanks >> Saisai >> >> On Wed, Mar 2, 2016 at 7:08 PM, Vinti Maheshwari <vinti.u...@gmail.com> >> wrote: >> >>> Hi All, >>> >>> I wanted to set *StorageLevel.MEMORY_AND_DISK_SER* in my >>> spark-streaming program as currently i am getting >>> MetadataFetchFailedException*. *I am not sure where i should pass >>> StorageLevel.MEMORY_AND_DISK, as it seems like createDirectStream >>> doesn't allow to pass that parameter. >>> >>> >>> val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, >>> StringDecoder]( >>> ssc, kafkaParams, topicsSet) >>> >>> >>> Full Error: >>> >>> *org.apache.spark.shuffle.MetadataFetchFailedException: Missing an >>> output location for shuffle 0* >>> at >>> org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:460) >>> at >>> org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:456) >>> at >>> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) >>> at >>> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) >>> at >>> scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) >>> at >>> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771) >>> at >>> org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.scala:456) >>> at >>> org.apache.spark.MapOutputTracker.getMapSizesByExecutorId(MapOutputTracker.scala:183) >>> at >>> org.apache.spark.shuffle.hash.HashShuffleReader.read(HashShuffleReader.scala:47) >>> at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:90) >>> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) >>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) >>> at >>> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) >>> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300) >>> at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:69) >>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:262) >>> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) >>> at org.apache.spark.scheduler.Task.run(Task.scala:88) >>> at >>> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) >>> at >>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) >>> at >>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) >>> at java.lang.Thread.run(Thread.java:745) >>> >>> ) >>> >>> Thanks, >>> ~Vinti >>> >>> >> >
