Same code of yours works for me as well
On Fri, Jun 26, 2015 at 8:02 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <deepuj...@gmail.com> wrote:
> Is that its not supported with Avro. Unlikely.
>
> On Fri, Jun 26, 2015 at 8:01 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <deepuj...@gmail.com>
> wrote:
>
>> 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 <
>> silvio.fior...@granturing.com> 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 <
>>> silvio.fior...@granturing.com> 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 <
>>>> silvio.fior...@granturing.com> 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 <
>>>>> ak...@sigmoidanalytics.com> 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, ÐΞ€ρ@Ҝ (๏̯͡๏) <deepuj...@gmail.com>
>>>>>> 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 <
>>>>>>> ak...@sigmoidanalytics.com> wrote:
>>>>>>>
>>>>>>>> Cool. :)
>>>>>>>> On 24 Jun 2015 23:44, "ÐΞ€ρ@Ҝ (๏̯͡๏)" <deepuj...@gmail.com> wrote:
>>>>>>>>
>>>>>>>>> Its running now.
>>>>>>>>>
>>>>>>>>> On Wed, Jun 24, 2015 at 10:45 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>> deepuj...@gmail.com> 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, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>>> deepuj...@gmail.com> 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://
>>>>>>>>>>> sparkdri...@phxdpehdc9dn2137.stratus.phx.ebay.com:47708/]
>>>>>>>>>>> swallowing exception during message send
>>>>>>>>>>> (akka.remote.RemoteTransportExceptionNoStackTrace)
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Wed, Jun 24, 2015 at 10:31 AM, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>>>> deepuj...@gmail.com> 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 <
>>>>>>>>>>>> ak...@sigmoidanalytics.com> 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, ÐΞ€ρ@Ҝ (๏̯͡๏) <
>>>>>>>>>>>>> deepuj...@gmail.com> 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
>>
>>
>
>
> --
> Deepak
>
>
--
Deepak
