Mailing-List: contact user-help@spark.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
In-Reply-To: 
 <CANu_wT6kNHgCj6gRrKJDzsR_RsoUCSA=aVYzGmoo1P25w37Qfw@mail.gmail.com>
References: 
 <CANu_wT4uLy6VDjFks1r+nubdOvZtieFPaqG97p4iaDt1g=qwBw@mail.gmail.com>
	<CA+AHuK=8NO7JJsv5dwBeLWgfDp=TO8smWWqh6yhcAHq6SMjNiQ@mail.gmail.com>
	<CANu_wT6kNHgCj6gRrKJDzsR_RsoUCSA=aVYzGmoo1P25w37Qfw@mail.gmail.com>
Date: Wed, 14 Oct 2015 11:05:57 -0700
Message-ID: 
 <CANu_wT4atsgp6VZFbzX7oLoFkkago=PByhqJAoOXYBG_BAvxyw@mail.gmail.com>
Subject: Re: Spark 1.5 java.net.ConnectException: Connection refused
From: Spark Newbie <sparknewbie1234@gmail.com>
To: Tathagata Das <tdas@databricks.com>
Cc: user <user@spark.apache.org>
Content-Type: multipart/alternative; boundary=001a11c3c3480d4de70522146a0e

--001a11c3c3480d4de70522146a0e
Content-Type: text/plain; charset=UTF-8

Is it slowing things down or blocking progress.
>> I didn't see slowing of processing, but I do see jobs aborted
consecutively for a period of 18 batches (5 minute batch intervals). So I
am worried about what happened to the records that these jobs were
processing.
Also, one more thing to mention is that the
StreamingListenerBatchCompleted.numRecords information shows all received
records as processed even if the batch/job failed. The processing time as
well shows as the same time it takes for a successful batch.
It seems like it is the numRecords which was the input to the batch
regardless of whether they were successfully processed or not.

On Wed, Oct 14, 2015 at 11:01 AM, Spark Newbie <sparknewbie1234@gmail.com>
wrote:

> I ran 2 different spark 1.5 clusters that have been running for more than
> a day now. I do see jobs getting aborted due to task retry's maxing out
> (default 4) due to ConnectionException. It seems like the executors die and
> get restarted and I was unable to find the root cause (same app code and
> conf used on spark 1.4.1 I don't see ConnectionException).
>
> Another question related to this, what happens to the kinesis records
> received when Job gets aborted? In Spark-1.5 and kinesis-asl-1.5 (which I
> am using) does the job gets resubmitted with the same received records? Or
> does the kinesis-asl library get those records again based on sequence
> numbers it tracks? It would good for me to understand the story around
> lossless processing of kinesis records in Spark-1.5 + kinesis-asl-1.5 when
> jobs are aborted. Any pointers or quick explanation would be very helpful.
>
>
> On Tue, Oct 13, 2015 at 4:04 PM, Tathagata Das <tdas@databricks.com>
> wrote:
>
>> Is this happening too often? Is it slowing things down or blocking
>> progress. Failures once in a while is part of the norm, and the system
>> should take care of itself.
>>
>> On Tue, Oct 13, 2015 at 2:47 PM, Spark Newbie <sparknewbie1234@gmail.com>
>> wrote:
>>
>>> Hi Spark users,
>>>
>>> I'm seeing the below exception in my spark streaming application. It
>>> happens in the first stage where the kinesis receivers receive records and
>>> perform a flatMap operation on the unioned Dstream. A coalesce step also
>>> happens as a part of that stage for optimizing the performance.
>>>
>>> This is happening on my spark 1.5 instance using kinesis-asl-1.5. When I
>>> look at the executor logs I do not see any exceptions indicating the root
>>> cause of why there is no connectivity on xxx.xx.xx.xxx:36684 or when did
>>> that service go down.
>>>
>>> Any help debugging this problem will be helpful.
>>>
>>> 15/10/13 16:36:07 ERROR shuffle.RetryingBlockFetcher: Exception while
>>> beginning fetch of 1 outstanding blocks
>>> java.io.IOException: Failed to connect to
>>> ip-xxx-xx-xx-xxx.ec2.internal/xxx.xx.xx.xxx:36684
>>>         at
>>> org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:193)
>>>         at
>>> org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:156)
>>>         at
>>> org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:88)
>>>         at
>>> org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:140)
>>>         at
>>> org.apache.spark.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:120)
>>>         at
>>> org.apache.spark.network.netty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:97)
>>>         at
>>> org.apache.spark.network.BlockTransferService.fetchBlockSync(BlockTransferService.scala:89)
>>>         at
>>> org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:595)
>>>         at
>>> org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:593)
>>>         at
>>> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
>>>         at
>>> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
>>>         at
>>> org.apache.spark.storage.BlockManager.doGetRemote(BlockManager.scala:593)
>>>         at
>>> org.apache.spark.storage.BlockManager.getRemote(BlockManager.scala:579)
>>>         at
>>> org.apache.spark.storage.BlockManager.get(BlockManager.scala:623)
>>>         at
>>> org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:44)
>>>         at org.apache.spark.rdd.RDD.iterator(RDD.scala:262)
>>>         at
>>> org.apache.spark.rdd.CoGroupedRDD$$anonfun$compute$2.apply(CoGroupedRDD.scala:139)
>>>         at
>>> org.apache.spark.rdd.CoGroupedRDD$$anonfun$compute$2.apply(CoGroupedRDD.scala:135)
>>>         at
>>> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
>>>         at scala.collection.immutable.List.foreach(List.scala:318)
>>>         at
>>> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
>>>         at
>>> org.apache.spark.rdd.CoGroupedRDD.compute(CoGroupedRDD.scala:135)
>>>         at
>>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
>>>         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:297)
>>>         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:297)
>>>         at
>>> org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:69)
>>>         at org.apache.spark.rdd.RDD.iterator(RDD.scala:262)
>>>         at
>>> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
>>>         at
>>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
>>>         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:297)
>>>         at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
>>>         at
>>> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
>>>         at
>>> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
>>>         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)
>>> Caused by: java.net.ConnectException: Connection refused:
>>> ip-xxx-xx-xx-xxx.ec2.internal/xxx.xx.xx.xxx:36684
>>>
>>> Thanks,
>>> Bharath
>>>
>>>
>>
>

--001a11c3c3480d4de70522146a0e
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div><div><div>Is it slowing things down or blocking progr=
ess.<br></div>&gt;&gt; I didn&#39;t see slowing of processing, but I do see=
 jobs aborted consecutively for a period of 18 batches (5 minute batch inte=
rvals). So I am worried about what happened to the records that these jobs =
were processing. <br></div>Also, one more thing to mention is that the <spa=
n style=3D"background-color:rgb(228,228,255)">StreamingListenerBatchComplet=
ed.numRecords</span> information shows all received records as processed ev=
en if the batch/job failed. The processing time as well shows as the same t=
ime it takes for a successful batch.=C2=A0 <br></div>It seems like it is th=
e numRecords which was the input to the batch regardless of whether they we=
re successfully processed or not.<br></div><div class=3D"gmail_extra"><br><=
div class=3D"gmail_quote">On Wed, Oct 14, 2015 at 11:01 AM, Spark Newbie <s=
pan dir=3D"ltr">&lt;<a href=3D"mailto:sparknewbie1234@gmail.com" target=3D"=
_blank">sparknewbie1234@gmail.com</a>&gt;</span> wrote:<br><blockquote clas=
s=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;border-left:1px #ccc solid;pad=
ding-left:1ex"><div dir=3D"ltr"><div><div>I ran 2 different spark 1.5 clust=
ers that have been running for more than a day now. I do see jobs getting a=
borted due to task retry&#39;s maxing out (default 4) due to ConnectionExce=
ption. It seems like the executors die and get restarted and I was unable t=
o find the root cause (same app code and conf used on spark 1.4.1 I don&#39=
;t see ConnectionException). <br><br></div>Another question related to this=
, what happens to the kinesis records received when Job gets aborted? In Sp=
ark-1.5 and kinesis-asl-1.5 (which I am using) does the job gets resubmitte=
d with the same received records? Or does the kinesis-asl library get those=
 records again based on sequence numbers it tracks? It would good for me to=
 understand the story around lossless processing of kinesis records in Spar=
k-1.5 + kinesis-asl-1.5 when jobs are aborted. Any pointers or quick explan=
ation would be very helpful.<br><br></div></div><div class=3D"HOEnZb"><div =
class=3D"h5"><div class=3D"gmail_extra"><br><div class=3D"gmail_quote">On T=
ue, Oct 13, 2015 at 4:04 PM, Tathagata Das <span dir=3D"ltr">&lt;<a href=3D=
"mailto:tdas@databricks.com" target=3D"_blank">tdas@databricks.com</a>&gt;<=
/span> wrote:<br><blockquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8=
ex;border-left:1px #ccc solid;padding-left:1ex"><div dir=3D"ltr">Is this ha=
ppening too often? Is it slowing things down or blocking progress. Failures=
 once in a while is part of the norm, and the system should take care of it=
self.</div><div><div><div class=3D"gmail_extra"><br><div class=3D"gmail_quo=
te">On Tue, Oct 13, 2015 at 2:47 PM, Spark Newbie <span dir=3D"ltr">&lt;<a =
href=3D"mailto:sparknewbie1234@gmail.com" target=3D"_blank">sparknewbie1234=
@gmail.com</a>&gt;</span> wrote:<br><blockquote class=3D"gmail_quote" style=
=3D"margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex"><div dir=
=3D"ltr"><div><div><div>Hi Spark users,<br><br></div>I&#39;m seeing the bel=
ow exception in my spark streaming application. It happens in the first sta=
ge where the kinesis receivers receive records and perform a flatMap operat=
ion on the unioned Dstream. A coalesce step also happens as a part of that =
stage for optimizing the performance. <br><br></div>This is happening on my=
 spark 1.5 instance using kinesis-asl-1.5. When I look at the executor logs=
 I do not see any exceptions indicating the root cause of why there is no c=
onnectivity on xxx.xx.xx.xxx:36684 or when did that service go down. <br><b=
r></div>Any help debugging this problem will be helpful.<br><div><div><div>=
<div><br>15/10/13 16:36:07 ERROR shuffle.RetryingBlockFetcher: Exception wh=
ile beginning fetch of 1 outstanding blocks<br>java.io.IOException: Failed =
to connect to ip-xxx-xx-xx-xxx.ec2.internal/xxx.xx.xx.xxx:36684<br>=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.network.client.Tra=
nsportClientFactory.createClient(TransportClientFactory.java:193)<br>=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.network.client.Tra=
nsportClientFactory.createClient(TransportClientFactory.java:156)<br>=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.network.netty.Nett=
yBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scal=
a:88)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.net=
work.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.=
java:140)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark=
.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:120)<=
br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.network.n=
etty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:=
97)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.netwo=
rk.BlockTransferService.fetchBlockSync(BlockTransferService.scala:89)<br>=
=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.storage.Bloc=
kManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:595)<br>=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.storage.BlockManager$=
$anonfun$doGetRemote$2.apply(BlockManager.scala:593)<br>=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0 at scala.collection.mutable.ResizableArray$class.f=
oreach(ResizableArray.scala:59)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=
=A0 at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)<b=
r>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.storage.Bl=
ockManager.doGetRemote(BlockManager.scala:593)<br>=C2=A0=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0 at org.apache.spark.storage.BlockManager.getRemote(Block=
Manager.scala:579)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apa=
che.spark.storage.BlockManager.get(BlockManager.scala:623)<br>=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.CacheManager.getOrComput=
e(CacheManager.scala:44)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at o=
rg.apache.spark.rdd.RDD.iterator(RDD.scala:262)<br>=C2=A0=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.CoGroupedRDD$$anonfun$compute$2.=
apply(CoGroupedRDD.scala:139)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=
 at org.apache.spark.rdd.CoGroupedRDD$$anonfun$compute$2.apply(CoGroupedRDD=
.scala:135)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at scala.collecti=
on.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scal=
a:772)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at scala.collection.im=
mutable.List.foreach(List.scala:318)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=
=A0=C2=A0 at scala.collection.TraversableLike$WithFilter.foreach(Traversabl=
eLike.scala:771)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apach=
e.spark.rdd.CoGroupedRDD.compute(CoGroupedRDD.scala:135)<br>=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.computeOrReadCheckp=
oint(RDD.scala:297)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.ap=
ache.spark.rdd.RDD.iterator(RDD.scala:264)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0 at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartiti=
onsRDD.scala:38)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apach=
e.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)<br>=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.iterator(RDD.scala:=
264)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.=
MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)<br>=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.computeOrReadCheckpoin=
t(RDD.scala:297)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apach=
e.spark.CacheManager.getOrCompute(CacheManager.scala:69)<br>=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.iterator(RDD.scala:=
262)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.=
MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)<br>=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.computeOrReadCheckpoin=
t(RDD.scala:297)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apach=
e.spark.rdd.RDD.iterator(RDD.scala:264)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0 at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitions=
RDD.scala:38)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.s=
park.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)<br>=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.rdd.RDD.iterator(RDD.scala:264=
)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.schedul=
er.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)<br>=C2=A0=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.scheduler.ShuffleMapTask.runTask(=
ShuffleMapTask.scala:41)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at o=
rg.apache.spark.scheduler.Task.run(Task.scala:88)<br>=C2=A0=C2=A0=C2=A0=C2=
=A0=C2=A0=C2=A0=C2=A0 at org.apache.spark.executor.Executor$TaskRunner.run(=
Executor.scala:214)<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at java.u=
til.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)<b=
r>=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0 at java.util.concurrent.Thread=
PoolExecutor$Worker.run(ThreadPoolExecutor.java:615)<br>=C2=A0=C2=A0=C2=A0=
=C2=A0=C2=A0=C2=A0=C2=A0 at java.lang.Thread.run(Thread.java:745)<br>Caused=
 by: java.net.ConnectException: Connection refused: ip-xxx-xx-xx-xxx.ec2.in=
ternal/xxx.xx.xx.xxx:36684<br><br></div><div>Thanks,<br></div><div>Bharath<=
br></div><div><br></div></div></div></div></div>
</blockquote></div><br></div>
</div></div></blockquote></div><br></div>
</div></div></blockquote></div><br></div>

--001a11c3c3480d4de70522146a0e--