kinesis batches hang after YARN automatic driver restart

[Hster Geguri <hs...@gmail.com>]

Hello Wonderful Sparks Peoples,

We are testing AWS Kinesis/Spark Streaming (1.5) failover behavior with
Hadoop/Yarn 2.6 and 2.71 and want to understand expected behavior.

When I manually kill a yarn application master/driver with a linux kill -9,
YARN will automatically relaunch another master that successfully reads in
the previous checkpoint.

However- more than half the time, the kinesis executors (5 second batches)
don't continue processing immediately.  I.e. batches of 0 events are queued
for  5-9 minutes before it starts reprocessing the stream again. When I
drill down to the current job which is hanging- it shows all stages/tasks
are complete. I would expect the automatically relaunched behavior to be
similar to as if I had manually done a resubmit with spark-submit where the
stream processing continues within a minute of launch.

Any input is highly appreciated.

Thanks much,
Heji

Re: kinesis batches hang after YARN automatic driver restart

[Hster Geguri <hs...@gmail.com>]

Hello Tathagata,

Thank you for responding. I have read your excellent article on Zero Data
Loss many many times.

The Spark Streaming screen shows KCL consistently pulling events from the
stream after half a minute as per usual which gets queued up.

It's always the first two batches (0 events each) that take 2-10 minutes to
process in the transform stage. When I drill down to the task, I see that
during the transform stage, a few out of the 25 tasks will take minutes to
complete while the rest take 1-3secs.

The executor task appears to be hung on this thread:

java.net.SocketInputStream.socketRead0(Native Method)
java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
java.net.SocketInputStream.read(SocketInputStream.java:170)
java.net.SocketInputStream.read(SocketInputStream.java:141)
sun.security.ssl.InputRecord.readFully(InputRecord.java:465)
sun.security.ssl.InputRecord.readV3Record(InputRecord.java:593)
sun.security.ssl.InputRecord.read(InputRecord.java:532)
sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:961)
sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:918)
sun.security.ssl.AppInputStream.read(AppInputStream.java:105)
org.apache.http.impl.io.AbstractSessionInputBuffer.read(AbstractSessionInputBuffer.java:198)
org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:178)
org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:137)
com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:72)
com.amazonaws.event.ProgressInputStream.read(ProgressInputStream.java:151)
com.fasterxml.jackson.core.json.UTF8StreamJsonParser.loadMore(UTF8StreamJsonParser.java:176)
com.fasterxml.jackson.core.base.ParserBase.loadMoreGuaranteed(ParserBase.java:408)
com.fasterxml.jackson.core.json.UTF8StreamJsonParser._finishString2(UTF8StreamJsonParser.java:2184)
com.fasterxml.jackson.core.json.UTF8StreamJsonParser._finishString(UTF8StreamJsonParser.java:2165)
com.fasterxml.jackson.core.json.UTF8StreamJsonParser.getText(UTF8StreamJsonParser.java:279)
com.amazonaws.transform.JsonUnmarshallerContextImpl.readCurrentJsonTokenValue(JsonUnmarshallerContextImpl.java:129)
com.amazonaws.transform.JsonUnmarshallerContextImpl.readText(JsonUnmarshallerContextImpl.java:123)
com.amazonaws.transform.SimpleTypeJsonUnmarshallers$ByteBufferJsonUnmarshaller.unmarshall(SimpleTypeJsonUnmarshallers.java:185)
com.amazonaws.services.kinesis.model.transform.RecordJsonUnmarshaller.unmarshall(RecordJsonUnmarshaller.java:58)
com.amazonaws.services.kinesis.model.transform.RecordJsonUnmarshaller.unmarshall(RecordJsonUnmarshaller.java:31)
com.amazonaws.transform.ListUnmarshaller.unmarshallJsonToList(ListUnmarshaller.java:93)
com.amazonaws.transform.ListUnmarshaller.unmarshall(ListUnmarshaller.java:43)
com.amazonaws.services.kinesis.model.transform.GetRecordsResultJsonUnmarshaller.unmarshall(GetRecordsResultJsonUnmarshaller.java:50)
com.amazonaws.services.kinesis.model.transform.GetRecordsResultJsonUnmarshaller.unmarshall(GetRecordsResultJsonUnmarshaller.java:31)
com.amazonaws.http.JsonResponseHandler.handle(JsonResponseHandler.java:106)
com.amazonaws.http.JsonResponseHandler.handle(JsonResponseHandler.java:42)
com.amazonaws.http.AmazonHttpClient.handleResponse(AmazonHttpClient.java:1072)
com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java:746)
com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:489)
com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:310)
com.amazonaws.services.kinesis.AmazonKinesisClient.invoke(AmazonKinesisClient.java:2490)
com.amazonaws.services.kinesis.AmazonKinesisClient.getRecords(AmazonKinesisClient.java:1142)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator$$anonfun$3.apply(KinesisBackedBlockRDD.scala:214)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator$$anonfun$3.apply(KinesisBackedBlockRDD.scala:214)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator.retryOrTimeout(KinesisBackedBlockRDD.scala:258)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator.getRecordsAndNextKinesisIterator(KinesisBackedBlockRDD.scala:213)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator.getRecords(KinesisBackedBlockRDD.scala:199)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator.getNext(KinesisBackedBlockRDD.scala:155)
org.apache.spark.streaming.kinesis.KinesisSequenceRangeIterator.getNext(KinesisBackedBlockRDD.scala:127)
org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71)
scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:396)
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:369)
org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:203)
org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:73)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
org.apache.spark.scheduler.Task.run(Task.scala:88)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
java.lang.Thread.run(Thread.java:745)


Sincerely,
Heji



On Tue, Nov 3, 2015 at 3:14 AM, Tathagata Das <td...@databricks.com> wrote:

> The Kinesis integration underneath uses the KCL libraries which takes a
> minute or so sometimes to spin up the threads and start getting data from
> Kinesis. That is under normal conditions. In your case, it could be
> happening that because of your killing and restarting, the restarted KCL
> may be taking a while to get new lease and start getting data again.
>
> On Mon, Nov 2, 2015 at 11:26 AM, Hster Geguri <
> hster.investigates@gmail.com> wrote:
>
>> Hello Wonderful Sparks Peoples,
>>
>> We are testing AWS Kinesis/Spark Streaming (1.5) failover behavior with
>> Hadoop/Yarn 2.6 and 2.71 and want to understand expected behavior.
>>
>> When I manually kill a yarn application master/driver with a linux kill
>> -9, YARN will automatically relaunch another master that successfully reads
>> in the previous checkpoint.
>>
>> However- more than half the time, the kinesis executors (5 second
>> batches) don't continue processing immediately.  I.e. batches of 0 events
>> are queued for  5-9 minutes before it starts reprocessing the stream again.
>> When I drill down to the current job which is hanging- it shows all
>> stages/tasks are complete. I would expect the automatically relaunched
>> behavior to be similar to as if I had manually done a resubmit with
>> spark-submit where the stream processing continues within a minute of
>> launch.
>>
>> Any input is highly appreciated.
>>
>> Thanks much,
>> Heji
>>
>
>

Re: kinesis batches hang after YARN automatic driver restart

[Tathagata Das <td...@databricks.com>]

The Kinesis integration underneath uses the KCL libraries which takes a
minute or so sometimes to spin up the threads and start getting data from
Kinesis. That is under normal conditions. In your case, it could be
happening that because of your killing and restarting, the restarted KCL
may be taking a while to get new lease and start getting data again.

On Mon, Nov 2, 2015 at 11:26 AM, Hster Geguri <hs...@gmail.com>
wrote:

> Hello Wonderful Sparks Peoples,
>
> We are testing AWS Kinesis/Spark Streaming (1.5) failover behavior with
> Hadoop/Yarn 2.6 and 2.71 and want to understand expected behavior.
>
> When I manually kill a yarn application master/driver with a linux kill
> -9, YARN will automatically relaunch another master that successfully reads
> in the previous checkpoint.
>
> However- more than half the time, the kinesis executors (5 second batches)
> don't continue processing immediately.  I.e. batches of 0 events are queued
> for  5-9 minutes before it starts reprocessing the stream again. When I
> drill down to the current job which is hanging- it shows all stages/tasks
> are complete. I would expect the automatically relaunched behavior to be
> similar to as if I had manually done a resubmit with spark-submit where the
> stream processing continues within a minute of launch.
>
> Any input is highly appreciated.
>
> Thanks much,
> Heji
>