We have a Spark (2.3.1) Streaming app running over EMR (5.16) consuming from a single shard of AWS Kinesis from the past 2 years. Recently (for the past 2 months), we are having random errors in the app. We tried to update to the latest version of Spark and EMR and the error was not solved.
The problem
Out of the blue, in any moment and uptime and with apparently no indicators in the RAM, CPU, Networking, the Receiver stops sending messages to Spark and we have this error in the logs:
18/09/11 18:20:00 ERROR ReceiverTracker: Deregistered receiver for stream 0: Error while storing block into Spark - java.util.concurrent.TimeoutException: Futures timed out after [30 seconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:201)
at org.apache.spark.streaming.receiver.WriteAheadLogBasedBlockHandler.storeBlock(ReceivedBlockHandler.scala:210)
at org.apache.spark.streaming.receiver.ReceiverSupervisorImpl.pushAndReportBlock(ReceiverSupervisorImpl.scala:158)
at org.apache.spark.streaming.receiver.ReceiverSupervisorImpl.pushArrayBuffer(ReceiverSupervisorImpl.scala:129)
at org.apache.spark.streaming.receiver.Receiver.store(Receiver.scala:133)
at org.apache.spark.streaming.kinesis.KinesisReceiver.org$apache$spark$streaming$kinesis$KinesisReceiver$$storeBlockWithRanges(KinesisReceiver.scala:306)
at org.apache.spark.streaming.kinesis.KinesisReceiver$GeneratedBlockHandler.onPushBlock(KinesisReceiver.scala:357)
at org.apache.spark.streaming.receiver.BlockGenerator.pushBlock(BlockGenerator.scala:297)
at org.apache.spark.streaming.receiver.BlockGenerator.org$apache$spark$streaming$receiver$BlockGenerator$$keepPushingBlocks(BlockGenerator.scala:269)
at org.apache.spark.streaming.receiver.BlockGenerator$$anon$1.run(BlockGenerator.scala:110)
The Spark app continue working without problems, but it doesn't have records to process.
The only thing related I found in the internet was this Github Issue https://github.com/awslabs/amazon-kinesis-client/issues/185
What I've tried
The final version of the configuration looks like this
[
{
"Classification": "capacity-scheduler",
"Properties": {
"yarn.scheduler.capacity.resource-calculator": "org.apache.hadoop.yarn.util.resource.DominantResourceCalculator"
}
},
{
"Classification": "spark-defaults",
"Properties": {
"spark.yarn.am.attemptFailuresValidityInterval": "1h",
"spark.yarn.maxAppAttempts": "4",
"spark.yarn.executor.failuresValidityInterval": "1h",
"spark.task.maxFailures": "8",
"spark.task.reaper.enabled": "true",
"spark.task.reaper.killTimeout": "120s",
"spark.metrics.conf": "metrics.properties",
"spark.metrics.namespace": "spark",
"spark.streaming.stopGracefullyOnShutdown": "true",
"spark.streaming.receiver.writeAheadLog.enable": "true",
"spark.streaming.receiver.writeAheadLog.closeFileAfterWrite": "true",
"spark.streaming.driver.writeAheadLog.closeFileAfterWrite": "true",
"spark.streaming.backpressure.enabled": "true",
"spark.streaming.backpressure.initialRate": "500",
"spark.dynamicAllocation.enabled": "false",
"spark.default.parallelism": "18",
"spark.driver.cores": "3",
"spark.driver.memory": "4G",
"spark.driver.memoryOverhead": "1024",
"spark.driver.extraJavaOptions": "-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseCompressedOops",
"spark.executor.instances": "3",
"spark.executor.cores": "3",
"spark.executor.memory": "4G",
"spark.executor.memoryOverhead": "1024",
"spark.executor.heartbeatInterval": "20s",
"spark.io.compression.codec": "zstd",
"spark.rdd.compress": "true",
"spark.executor.extraJavaOptions": "-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseCompressedOops",
"spark.python.worker.memory": "640m",
"spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version": "2",
"spark.hadoop.mapreduce.fileoutputcommitter.cleanup-failures.ignored": "true",
"spark.network.timeout": "240s",
"spark.locality.wait": "0s",
"spark.eventLog.enabled": "false"
}
},
{
"Classification": "emrfs-site",
"Properties": {
"fs.s3.consistent.metadata.tableName": "HiddenName",
"fs.s3.consistent": "true",
"fs.s3.consistent.retryCount": "1",
"fs.s3.consistent.retryPeriodSeconds": "2",
"fs.s3.consistent.retryPolicyType": "fixed",
"fs.s3.consistent.throwExceptionOnInconsistency": "false"
}
}
]
Nothing there helped.
Workarounds To have the Receiver sending messages again, I have to go to the Spark UI, Jobs Tab and kill the active Job (I can't upload images because I don't have enough reputation). That triggers a new app and the Receiver starts receiving sending records to Spark again. Of course, this is not a solution for a Production System.
Programmatic workaround
I wrote a StreamingListener that captures the onReceiverStopped and onReceiverError. When that happens it creates a signal (external file) that is checked everytime ssc.awaitTerminationOrTimeout(check_interval) is processed. If the signal was created, the app raises an Exception and the app is killed. This generates a new appattempt and a new Streaming App runs. The problem with this, 1st is that it doesn't fix the original problem and 2nd it looks like after the first appattempt the StreamingListener is not registered anymore, so I can't recover the App.