Re: Spark streaming: StorageLevel.MEMORY_AND_DISK_SER setting for KafkaUtils.createDirectStream
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 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 > 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 >> 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 >>> >>> >> >
Re: Spark streaming: StorageLevel.MEMORY_AND_DISK_SER setting for KafkaUtils.createDirectStream
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 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 > 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 >> >> >
Re: Spark streaming: StorageLevel.MEMORY_AND_DISK_SER setting for KafkaUtils.createDirectStream
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 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 > >
Spark streaming: StorageLevel.MEMORY_AND_DISK_SER setting for KafkaUtils.createDirectStream
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