My imports:
import org.apache.avro.generic.GenericData
import org.apache.avro.generic.GenericRecord
import org.apache.avro.mapred.AvroKey
import org.apache.avro.Schema
import org.apache.hadoop.io.NullWritable
import org.apache.avro.mapreduce.AvroKeyInputFormat
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.Text
def readGenericRecords(sc: SparkContext, inputDir: String, startDate:
Date, endDate: Date) = {
val path = getInputPaths(inputDir, startDate, endDate)
val hadoopConfiguration = new Configuration(sc.hadoopConfiguration)
hadoopConfiguration.set("mapreduce.input.fileinputformat.split.maxsize",
"67108864")
sc.newAPIHadoopFile[AvroKey[GenericRecord], NullWritable,
AvroKeyInputFormat[GenericRecord]](path + "/*.avro")
}
I need to read Avro datasets and am using strings instead of constant from
InputFormat class.
When i click on any hadoop dependency from eclipse, i see they point to
hadoop 2.2.x jars.
On Fri, Jun 26, 2015 at 7:44 AM, Silvio Fiorito <
[email protected]> wrote:
> Make sure you’re importing the right namespace for Hadoop v2.0. This is
> what I tried:
>
> import org.apache.hadoop.io.{LongWritable, Text}
> import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat,
> TextInputFormat}
>
> val hadoopConf = new org.apache.hadoop.conf.Configuration()
> hadoopConf.setLong(FileInputFormat.SPLIT_MAXSIZE, 2048)
>
> val input = sc.newAPIHadoopFile(
> "README.md",
> classOf[TextInputFormat],
> classOf[LongWritable],
> classOf[Text],
> hadoopConf).map(_._2.toString())
>
> println(input.partitions.size)
>
> input.
> flatMap(_.split(" ")).
> filter(_.length > 0).
> map((_, 1)).
> reduceByKey(_ + _).
> coalesce(1).
> sortBy(_._2, false).
> take(10).
> foreach(println)
>
>
> From: "ÐΞ€ρ@Ҝ (๏̯͡๏)"
> Date: Friday, June 26, 2015 at 10:18 AM
> To: Silvio Fiorito
> Cc: user
> Subject: Re:
>
> All these throw compilation error at newAPIHadoopFile
>
> 1)
>
> val hadoopConfiguration = new Configuration()
>
> hadoopConfiguration.set(
> "mapreduce.input.fileinputformat.split.maxsize", "67108864")
>
> sc.newAPIHadoopFile[AvroKey, NullWritable, AvroKeyInputFormat](path +
> "/*.avro", classOf[AvroKey], classOf[NullWritable],
> classOf[AvroKeyInputFormat], hadoopConfiguration)
>
> 2)
>
> val hadoopConfiguration = new Configuration()
>
> hadoopConfiguration.set(
> "mapreduce.input.fileinputformat.split.maxsize", "67108864")
>
> sc.newAPIHadoopFile[AvroKey, NullWritable, AvroKeyInputFormat](path +
> "/*.avro", classOf[AvroKey[GenericRecord]], classOf[NullWritable],
> classOf[AvroKeyInputFormat[GenericRecord]],hadoopConfiguration)
>
> 3)
>
> val hadoopConfiguration = new Configuration()
>
> hadoopConfiguration.set(
> "mapreduce.input.fileinputformat.split.maxsize", "67108864")
>
> sc.newAPIHadoopFile[AvroKey[GenericRecord], NullWritable,
> AvroKeyInputFormat[GenericRecord]](path + "/*.avro",
> classOf[AvroKey[GenericRecord]], classOf[NullWritable],
> classOf[AvroKeyInputFormat[GenericRecord]], hadoopConfiguration)
>
> Error:
>
> [ERROR]
> /Users/dvasthimal/ebay/projects/ep/ep-spark/src/main/scala/com/ebay/ep/poc/spark/reporting/process/util/DataUtil.scala:37:
> error: overloaded method value newAPIHadoopFile with alternatives:
>
> [INFO] (path: String,fClass:
> Class[org.apache.avro.mapreduce.AvroKeyInputFormat[org.apache.avro.generic.GenericRecord]],kClass:
> Class[org.apache.avro.mapred.AvroKey[org.apache.avro.generic.GenericRecord]],vClass:
> Class[org.apache.hadoop.io.NullWritable],conf:
> org.apache.hadoop.conf.Configuration)org.apache.spark.rdd.RDD[(org.apache.avro.mapred.AvroKey[org.apache.avro.generic.GenericRecord],
> org.apache.hadoop.io.NullWritable)] <and>
>
> [INFO] (path: String)(implicit km:
> scala.reflect.ClassTag[org.apache.avro.mapred.AvroKey[org.apache.avro.generic.GenericRecord]],
> implicit vm: scala.reflect.ClassTag[org.apache.hadoop.io.NullWritable],
> implicit fm:
> scala.reflect.ClassTag[org.apache.avro.mapreduce.AvroKeyInputFormat[org.apache.avro.generic.GenericRecord]])org.apache.spark.rdd.RDD[(org.apache.avro.mapred.AvroKey[org.apache.avro.generic.GenericRecord],
> org.apache.hadoop.io.NullWritable)]
>
> [INFO] cannot be applied to (String,
> Class[org.apache.avro.mapred.AvroKey[org.apache.avro.generic.GenericRecord]],
> Class[org.apache.hadoop.io.NullWritable],
> Class[org.apache.avro.mapreduce.AvroKeyInputFormat[org.apache.avro.generic.GenericRecord]],
> org.apache.hadoop.conf.Configuration)
>
> [INFO] sc.newAPIHadoopFile[AvroKey[GenericRecord], NullWritable,
> AvroKeyInputFormat[GenericRecord]](path + "/*.avro",
> classOf[AvroKey[GenericRecord]], classOf[NullWritable],
> classOf[AvroKeyInputFormat[GenericRecord]], hadoopConfiguration)
>
>
>
> On Thu, Jun 25, 2015 at 4:14 PM, Silvio Fiorito <
> [email protected]> wrote:
>
>> Ok, in that case I think you can set the max split size in the Hadoop
>> config object, using the FileInputFormat.SPLIT_MAXSIZE config parameter.
>>
>> Again, I haven’t done this myself, but looking through the Spark
>> codebase here:
>> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/SparkContext.scala#L1053
>>
>> And the HDFS FileInputFormat implementation, that seems like a good
>> option to try.
>>
>> You should be able to call conf.setLong(FileInputFormat.SPLIT_MAXSIZE,
>> max).
>>
>> I hope that helps!
>>
>> From: "ÐΞ€ρ@Ҝ (๏̯͡๏)"
>> Date: Thursday, June 25, 2015 at 5:49 PM
>> To: Silvio Fiorito
>> Cc: user
>> Subject: Re:
>>
>> I use
>>
>> sc.newAPIHadoopFile[AvroKey[GenericRecord], NullWritable,
>> AvroKeyInputFormat[GenericRecord]](path + "/*.avro")
>>
>>
>>
>> https://spark.apache.org/docs/1.3.1/api/java/org/apache/spark/SparkContext.html#newAPIHadoopFile(java.lang.String,
>> java.lang.Class, java.lang.Class, java.lang.Class,
>> org.apache.hadoop.conf.Configuration)
>>
>> Does not seem to have that partition option.
>>
>> On Thu, Jun 25, 2015 at 12:24 PM, Silvio Fiorito <
>> [email protected]> wrote:
>>
>>> Hi Deepak,
>>>
>>> Have you tried specifying the minimum partitions when you load the
>>> file? I haven’t tried that myself against HDFS before, so I’m not sure if
>>> it will affect data locality. Ideally not, it should still maintain data
>>> locality but just more partitions. Once your job runs, you can check in the
>>> Spark tasks web UI to ensure they’re all Node local.
>>>
>>> val details = sc.textFile(“hdfs://….”, 500)
>>>
>>> If you’re using something other than text file you can also specify
>>> minimum partitions when using sc.hadoopFile.
>>>
>>> Thanks,
>>> Silvio
>>>
>>> From: "ÐΞ€ρ@Ҝ (๏̯͡๏)"
>>> Date: Thursday, June 25, 2015 at 3:10 PM
>>> To: Akhil Das
>>> Cc: user
>>> Subject: Re:
>>>
>>> How can i increase the number of tasks from 174 to 500 without
>>> running repartition.
>>>
>>> The input size is 512.0 MB (hadoop) / 4159106. Can this be reduced to
>>> 64 MB so as to increase the number of tasks. Similar to split size that
>>> increases the number of mappers in Hadoop M/R.
>>>
>>> On Thu, Jun 25, 2015 at 12:06 AM, Akhil Das <[email protected]>
>>> wrote:
>>>
>>>> Look in the tuning section
>>>> <https://spark.apache.org/docs/latest/tuning.html>, also you need to
>>>> figure out whats taking time and where's your bottleneck etc. If everything
>>>> is tuned properly, then you will need to throw more cores :)
>>>>
>>>> Thanks
>>>> Best Regards
>>>>
>>>> On Thu, Jun 25, 2015 at 12:19 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <[email protected]>
>>>> wrote:
>>>>
>>>>> Its taking an hour and on Hadoop it takes 1h 30m, is there a way to
>>>>> make it run faster ?
>>>>>
>>>>> On Wed, Jun 24, 2015 at 11:39 AM, Akhil Das <
>>>>> [email protected]> wrote:
>>>>>
>>>>>> Cool. :)
>>>>>> On 24 Jun 2015 23:44, "ÐΞ€ρ@Ҝ (๏̯͡๏)" <[email protected]> wrote:
>>>>>>
>>>>>>> Its running now.
>>>>>>>
>>>>>>> On Wed, Jun 24, 2015 at 10:45 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <[email protected]
>>>>>>> > wrote:
>>>>>>>
>>>>>>>> Now running with
>>>>>>>>
>>>>>>>> *--num-executors 9973 --driver-memory 14g --driver-java-options
>>>>>>>> "-XX:MaxPermSize=512M -Xmx4096M -Xms4096M" --executor-memory 14g
>>>>>>>> --executor-cores 1*
>>>>>>>>
>>>>>>>>
>>>>>>>> On Wed, Jun 24, 2015 at 10:34 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>> [email protected]> wrote:
>>>>>>>>
>>>>>>>>> There are multiple of these
>>>>>>>>>
>>>>>>>>> 1)
>>>>>>>>> 15/06/24 09:53:37 ERROR executor.Executor: Exception in task 443.0
>>>>>>>>> in stage 3.0 (TID 1767)
>>>>>>>>> java.lang.OutOfMemoryError: GC overhead limit exceeded
>>>>>>>>> at
>>>>>>>>> sun.reflect.GeneratedSerializationConstructorAccessor1327.newInstance(Unknown
>>>>>>>>> Source)
>>>>>>>>> at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
>>>>>>>>> at
>>>>>>>>> org.objenesis.instantiator.sun.SunReflectionFactoryInstantiator.newInstance(SunReflectionFactoryInstantiator.java:56)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.newInstance(Kryo.java:1065)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.create(FieldSerializer.java:228)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:217)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:134)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:17)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
>>>>>>>>> at
>>>>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
>>>>>>>>> at
>>>>>>>>> com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:42)
>>>>>>>>> at
>>>>>>>>> com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:33)
>>>>>>>>> at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.serializer.KryoDeserializationStream.readObject(KryoSerializer.scala:138)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.serializer.DeserializationStream$$anon$1.getNext(Serializer.scala:133)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
>>>>>>>>> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
>>>>>>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>>>> at
>>>>>>>>> org.apache.spark.util.collection.ExternalAppendOnlyMap.insertAll(ExternalAppendOnlyMap.scala:125)
>>>>>>>>> 15/06/24 09:53:37 ERROR actor.ActorSystemImpl: exception on
>>>>>>>>> LARS? timer thread
>>>>>>>>>
>>>>>>>>> 2)
>>>>>>>>> 15/06/24 09:53:37 ERROR actor.ActorSystemImpl: exception on LARS?
>>>>>>>>> timer thread
>>>>>>>>> java.lang.OutOfMemoryError: GC overhead limit exceeded
>>>>>>>>> at
>>>>>>>>> akka.dispatch.AbstractNodeQueue.<init>(AbstractNodeQueue.java:22)
>>>>>>>>> at
>>>>>>>>> akka.actor.LightArrayRevolverScheduler$TaskQueue.<init>(Scheduler.scala:443)
>>>>>>>>> at
>>>>>>>>> akka.actor.LightArrayRevolverScheduler$$anon$8.nextTick(Scheduler.scala:409)
>>>>>>>>> at
>>>>>>>>> akka.actor.LightArrayRevolverScheduler$$anon$8.run(Scheduler.scala:375)
>>>>>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>>>>>> 3)
>>>>>>>>> # java.lang.OutOfMemoryError: GC overhead limit exceeded
>>>>>>>>> # -XX:OnOutOfMemoryError="kill %p"
>>>>>>>>> # Executing /bin/sh -c "kill 20674"...
>>>>>>>>> [ERROR] [06/24/2015 09:53:37.590] [Executor task launch worker-5]
>>>>>>>>> [akka.tcp://
>>>>>>>>> [email protected]:47708/]
>>>>>>>>> swallowing exception during message send
>>>>>>>>> (akka.remote.RemoteTransportExceptionNoStackTrace)
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On Wed, Jun 24, 2015 at 10:31 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>> [email protected]> wrote:
>>>>>>>>>
>>>>>>>>>> I see this
>>>>>>>>>>
>>>>>>>>>> java.lang.OutOfMemoryError: GC overhead limit exceeded
>>>>>>>>>> at java.util.Arrays.copyOfRange(Arrays.java:2694)
>>>>>>>>>> at java.lang.String.<init>(String.java:203)
>>>>>>>>>> at java.lang.StringBuilder.toString(StringBuilder.java:405)
>>>>>>>>>> at java.io.UnixFileSystem.resolve(UnixFileSystem.java:108)
>>>>>>>>>> at java.io.File.<init>(File.java:367)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.storage.DiskBlockManager.getFile(DiskBlockManager.scala:81)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.storage.DiskBlockManager.getFile(DiskBlockManager.scala:84)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.shuffle.IndexShuffleBlockManager.getIndexFile(IndexShuffleBlockManager.scala:60)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.shuffle.IndexShuffleBlockManager.getBlockData(IndexShuffleBlockManager.scala:107)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.storage.BlockManager.getBlockData(BlockManager.scala:304)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:57)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:57)
>>>>>>>>>> at
>>>>>>>>>> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>>>>>>>>>> at
>>>>>>>>>> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>>>>>>>>>> at
>>>>>>>>>> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
>>>>>>>>>> at
>>>>>>>>>> scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
>>>>>>>>>> at
>>>>>>>>>> scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
>>>>>>>>>> at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:57)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:124)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:97)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:91)
>>>>>>>>>> at
>>>>>>>>>> org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:44)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319)
>>>>>>>>>> at
>>>>>>>>>> io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319)
>>>>>>>>>> at
>>>>>>>>>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:163)
>>>>>>>>>> at
>>>>>>>>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333)
>>>>>>>>>>
>>>>>>>>>> On Wed, Jun 24, 2015 at 7:16 AM, Akhil Das <
>>>>>>>>>> [email protected]> wrote:
>>>>>>>>>>
>>>>>>>>>>> Can you look a bit more in the error logs? It could be getting
>>>>>>>>>>> killed because of OOM etc. One thing you can try is to set the
>>>>>>>>>>> spark.shuffle.blockTransferService to nio from netty.
>>>>>>>>>>>
>>>>>>>>>>> Thanks
>>>>>>>>>>> Best Regards
>>>>>>>>>>>
>>>>>>>>>>> On Wed, Jun 24, 2015 at 5:46 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>>>> [email protected]> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> I have a Spark job that has 7 stages. The first 3 stage
>>>>>>>>>>>> complete and the fourth stage beings (joins two RDDs). This stage
>>>>>>>>>>>> has
>>>>>>>>>>>> multiple task failures all the below exception.
>>>>>>>>>>>>
>>>>>>>>>>>> Multiple tasks (100s) of them get the same exception with
>>>>>>>>>>>> different hosts. How can all the host suddenly stop responding
>>>>>>>>>>>> when few
>>>>>>>>>>>> moments ago 3 stages ran successfully. If I re-run the three
>>>>>>>>>>>> stages will
>>>>>>>>>>>> again run successfully. I cannot think of it being a cluster issue.
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> Any suggestions ?
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> Spark Version : 1.3.1
>>>>>>>>>>>>
>>>>>>>>>>>> Exception:
>>>>>>>>>>>>
>>>>>>>>>>>> org.apache.spark.shuffle.FetchFailedException: Failed to connect
>>>>>>>>>>>> to HOST
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.shuffle.hash.BlockStoreShuffleFetcher$.org$apache$spark$shuffle$hash$BlockStoreShuffleFetcher$$unpackBlock$1(BlockStoreShuffleFetcher.scala:67)
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.shuffle.hash.BlockStoreShuffleFetcher$$anonfun$3.apply(BlockStoreShuffleFetcher.scala:83)
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.shuffle.hash.BlockStoreShuffleFetcher$$anonfun$3.apply(BlockStoreShuffleFetcher.scala:83)
>>>>>>>>>>>> at
>>>>>>>>>>>> scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
>>>>>>>>>>>> at
>>>>>>>>>>>> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>>>>>>> at
>>>>>>>>>>>> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>>>>>>> at
>>>>>>>>>>>> org.apache.spark.util.collection.ExternalAppendOnlyMap.insertAll(ExternalAppendOnlyMap.scala:125)
>>>>>>>>>>>> at org.apache.sp
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> --
>>>>>>>>>>>> Deepak
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> --
>>>>>>>>>> Deepak
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> --
>>>>>>>>> Deepak
>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> --
>>>>>>>> Deepak
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Deepak
>>>>>>>
>>>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Deepak
>>>>>
>>>>>
>>>>
>>>
>>>
>>> --
>>> Deepak
>>>
>>>
>>
>>
>> --
>> Deepak
>>
>>
>
>
> --
> Deepak
>
>
--
Deepak
