Hello,
We are building a Spark Streaming application that listens to a directory
on HDFS, and uses the SolrJ library to send newly detected files to a Solr
server. When we put 10.000 files to the directory it is listening to, it
starts to process them by sending the files to our Solr server but after
about a few thousand files the Spark Streaming application dies.
Before the application dies, It gives some TimeoutException errors related
to Akka, such as:
util.AkkaUtils: Error sending message in 1 attempts
java.util.concurrent.TimeoutException: Futures timed out after [30
seconds]
akka.pattern.AskTimeoutException: Timed out
Any ideas on how to deal with this? Should we add/change/tweak some Spark
configuration parameters?
We're using Spark 1.2.0 on a YARN cluster, and we're giving 4 cores and 2GB
of memory to that application when invoking it via spark-submit command.
Below you can read the last few lines of the log file, showing what our
Spark Streaming application logged just before it died:
15/02/20 13:28:25 INFO rdd.NewHadoopRDD: Input split:
hdfs://node01.demo.hadoop:8020/user/bjorn/spark-belga/dbpedia-translator-out/2fdf95f1-67d6-40b7-9345-fe129e38a2d9.json:0+2620
15/02/20 13:28:25 INFO broadcast.TorrentBroadcast: Started reading
broadcast variable 3004
15/02/20 13:28:32 INFO storage.MemoryStore: ensureFreeSpace(20996) called
with curMem=31171148, maxMem=1111794647
15/02/20 13:28:32 INFO storage.MemoryStore: Block broadcast_3004_piece0
stored as bytes in memory (estimated size 20.5 KB, free 1030.5 MB)
15/02/20 13:28:33 INFO storage.BlockManagerMaster: Updated info of block
broadcast_3004_piece0
15/02/20 13:28:33 INFO broadcast.TorrentBroadcast: Reading broadcast
variable 3004 took 7897 ms
15/02/20 13:28:33 INFO storage.MemoryStore: ensureFreeSpace(347363) called
with curMem=31192144, maxMem=1111794647
15/02/20 13:28:33 INFO storage.MemoryStore: Block broadcast_3004 stored as
values in memory (estimated size 339.2 KB, free 1030.2 MB)
15/02/20 13:28:33 INFO storage.MemoryStore: ensureFreeSpace(2627) called
with curMem=31539507, maxMem=1111794647
15/02/20 13:28:33 INFO storage.MemoryStore: Block rdd_3659_3 stored as
bytes in memory (estimated size 2.6 KB, free 1030.2 MB)
15/02/20 13:28:34 INFO storage.BlockManagerMaster: Updated info of block
rdd_3659_3
15/02/20 13:28:34 INFO impl.HttpClientUtil: Creating new http client,
config:maxConnections=128&maxConnectionsPerHost=32&followRedirects=false
15/02/20 13:28:36 INFO storage.MemoryStore: ensureFreeSpace(5) called with
curMem=31542134, maxMem=1111794647
15/02/20 13:28:36 INFO storage.MemoryStore: Block rdd_3660_3 stored as
bytes in memory (estimated size 5.0 B, free 1030.2 MB)
15/02/20 13:28:40 INFO storage.BlockManagerMaster: Updated info of block
rdd_3660_3
15/02/20 13:28:40 INFO executor.Executor: Finished task 3.0 in stage 245.0
(TID 3455). 2516 bytes result sent to driver
15/02/20 13:29:07 WARN util.AkkaUtils: Error sending message in 1 attempts
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 scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at
scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at org.apache.spark.util.AkkaUtils$.askWithReply(AkkaUtils.scala:187)
at org.apache.spark.executor.Executor$$anon$1.run(Executor.scala:398)
15/02/20 13:29:08 ERROR executor.CoarseGrainedExecutorBackend: Driver
Disassociated [akka.tcp://sparkexecu...@node08.demo.hadoop:48042] ->
[akka.tcp://sparkdri...@node07.demo.hadoop:56535] disassociated! Shutting
down.
LogType: stdout
LogLength: 0
Log Contents:
Container: container_1422006251277_0837_01_000004 on node08.demo.hadoop_8041
==============================================================================
LogType: stderr
LogLength: 2952
Log Contents:
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in
[jar:file:/mnt/disk1/cm/parcels/CDH-5.3.0-1.cdh5.3.0.p0.30/jars/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in
[jar:file:/mnt/disk3/yarn/nm/usercache/bjorn/filecache/354/bigcontent-1.0-SNAPSHOT.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
15/02/20 13:29:26 INFO executor.CoarseGrainedExecutorBackend: Registered
signal handlers for [TERM, HUP, INT]
15/02/20 13:29:27 INFO spark.SecurityManager: Changing view acls to:
yarn,bjorn
15/02/20 13:29:27 INFO spark.SecurityManager: Changing modify acls to:
yarn,bjorn
15/02/20 13:29:27 INFO spark.SecurityManager: SecurityManager:
authentication disabled; ui acls disabled; users with view permissions:
Set(yarn, bjorn); users with modify permissions: Set(yarn, bjorn)
15/02/20 13:29:28 INFO slf4j.Slf4jLogger: Slf4jLogger started
15/02/20 13:29:28 INFO Remoting: Starting remoting
15/02/20 13:29:29 INFO Remoting: Remoting started; listening on addresses
:[akka.tcp://driverpropsfetc...@node08.demo.hadoop:60995]
15/02/20 13:29:29 INFO Remoting: Remoting now listens on addresses:
[akka.tcp://driverpropsfetc...@node08.demo.hadoop:60995]
15/02/20 13:29:29 INFO util.Utils: Successfully started service
'driverPropsFetcher' on port 60995.
15/02/20 13:29:59 WARN security.UserGroupInformation:
PriviledgedActionException as:bjorn (auth:SIMPLE)
cause:akka.pattern.AskTimeoutException: Timed out
Exception in thread "main" java.lang.reflect.UndeclaredThrowableException
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1655)
at
org.apache.spark.deploy.SparkHadoopUtil.runAsSparkUser(SparkHadoopUtil.scala:59)
at
org.apache.spark.executor.CoarseGrainedExecutorBackend$.run(CoarseGrainedExecutorBackend.scala:115)
at
org.apache.spark.executor.CoarseGrainedExecutorBackend$.main(CoarseGrainedExecutorBackend.scala:161)
at
org.apache.spark.executor.CoarseGrainedExecutorBackend.main(CoarseGrainedExecutorBackend.scala)
Caused by: akka.pattern.AskTimeoutException: Timed out
at
akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:334)
at akka.actor.Scheduler$$anon$11.run(Scheduler.scala:118)
at
scala.concurrent.Future$InternalCallbackExecutor$.scala$concurrent$Future$InternalCallbackExecutor$$unbatchedExecute(Future.scala:694)
at
scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:691)
at
akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(Scheduler.scala:455)
at
akka.actor.LightArrayRevolverScheduler$$anon$12.executeBucket$1(Scheduler.scala:407)
at
akka.actor.LightArrayRevolverScheduler$$anon$12.nextTick(Scheduler.scala:411)
at
akka.actor.LightArrayRevolverScheduler$$anon$12.run(Scheduler.scala:363)
at java.lang.Thread.run(Thread.java:745)
Kind regards,
Emre Sevinç
http://www.bigindustries.be/
