Nice write-up... very helpful!
-----Original Message----- From: Tim Smith [mailto:secs...@gmail.com] Sent: Wednesday, September 17, 2014 1:11 PM Cc: spark users Subject: Re: Stable spark streaming app I don't have anything in production yet but I now at least have a stable (running for more than 24 hours) streaming app. Earlier, the app would crash for all sorts of reasons. Caveats/setup: - Spark 1.0.0 (I have no input flow control unlike Spark 1.1) - Yarn for RM - Input and Output to Kafka - CDH 5.1 - 11 node cluster with 32-cores and 48G max container size for each node (Yarn managed) - 5 partition Kafka topic - both in and out - Roughly, an average of 25k messages per second - App written in Scala (warning: I am a Scala noob) Few things I had to add/tweak to get the app to be stable: - The executor JVMs did not have any GC options set, by default. This might be more of a CDH issue. I noticed that while the Yarn container and other Spark ancillary tasks had GC options set at launch but none for the executors. So I played with different GC options and this worked best: SPARK_JAVA_OPTS="-XX:MaxPermSize=512m -XX:NewSize=1024m -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=70 -XX:+AggressiveHeap -XX:MaxHeapFreeRatio=70 -verbosegc -XX:+PrintGCDetails" I tried G1GC but for some reason it just didn't work. I am not a Java programmer or expert so my conclusion is purely trial and error based. The GC logs, with these flags, go to the "stdout" file in the Yarn container logs on each node/worker. You can set SPARK_JAVA_OPTS in spark-env.sh on the driver node and Yarn will respect these. On CDH/CM specifically, even though you don't run Spark as a service (since you are using Yarn for RM), you can goto "Spark Client Advanced Configuration Snippet (Safety Valve) for spark-conf/spark-env.sh" and set SPARK_JAVA_OPTS there. - Set these two params - "spark.yarn.executor.memoryOverhead" "spark.yarn.driver.memoryOverhead". Earlier, my app would get killed because the executors running the kafka receivers would get killed by Yarn for over utilization of memory. Now, these are my memory settings (I will paste the entire app launch params later in the email): --driver-memory 2G \ --executor-memory 16G \ --spark.yarn.executor.memoryOverhead 4096 \ --spark.yarn.driver.memoryOverhead 1024 \ Your total executor JVM will consume "executor-memory" minus "spark.yarn.executor.memoryOverhead" so you should see each executor JVM consuming no more than 12G, in this case. Here is how I launch my app: run=`date +"%m-%d-%YT%T"`; \ nohup spark-submit --class myAwesomeApp \ --master yarn myawesomeapp.jar \ --jars spark-streaming-kafka_2.10-1.0.0.jar,kafka_2.10-0.8.1.1.jar,zkclient-0.3.jar,metrics-core-2.2.0.jar,json4s-jackson_2.10-3.2.10.jar \ --driver-memory 2G \ --executor-memory 16G \ --executor-cores 16 \ --num-executors 10 \ --spark.serializer org.apache.spark.serializer.KryoSerializer \ --spark.rdd.compress true \ --spark.io.compression.codec org.apache.spark.io.SnappyCompressionCodec \ --spark.akka.threads 64 \ --spark.akka.frameSize 500 \ --spark.task.maxFailures 64 \ --spark.scheduler.mode FAIR \ --spark.yarn.executor.memoryOverhead 4096 \ --spark.yarn.driver.memoryOverhead 1024 \ --spark.shuffle.consolidateFiles true \ --spark.default.parallelism 528 \ >logs/normRunLog-$run.log \ 2>logs/normRunLogError-$run.log & \ echo $! > logs/current-run.pid Some code optimizations (or, goof ups that I fixed). I did not scientifically measure the impact of each but I think they helped: - Made all my classes and objects serializable and then use Kryo (as you see above) - I map one receive task for each kafka partition - Instead of doing a "union" on all the incoming streams and then repartition() I now repartition() each incoming stream and process them separately. I believe this reduces shuffle. - Reduced number of repartitions. I was doing 128 after doing a "union" on all incoming dStreams. I now repartition each of the five streams separately (in a loop) to 24. - For each RDD, I set storagelevel to "MEMORY_AND_DISK_SER" - Process data per partition instead of per RDD: dataout.foreachRDD( rdd => rdd.foreachPartition(rec => { myFunc(rec) }) ) - Specific to kafka: when I create "new Producer", make sure I "close" it else I had a ton of "too many files open" errors :) - Use immutable objects as far as possible. If I use mutable objects within a method/class then I turn them into immutable before passing onto another class/method. - For logging, create a LogService object that I then use for other object/class declarations. Once instantiated, I can make "logInfo" calls from within other Objects/Methods/Classes and output goes to the "stderr" file in the Yarn container logs. Good for debugging stream processing logic. Currently, my processing delay is lower than my dStream time window so all is good. I get a ton of these errors in my driver logs: 14/09/16 21:17:40 ERROR LiveListenerBus: Listener JobProgressListener threw an exception These seem related to: https://issues.apache.org/jira/browse/SPARK-2316 Best I understand and have been told, this does not affect data integrity but may cause un-necessary recomputes. Hope this helps, Tim On Wed, Sep 17, 2014 at 8:30 AM, Soumitra Kumar <kumar.soumi...@gmail.com> wrote: > Hmm, no response to this thread! > > Adding to it, please share experiences of building an enterprise grade > product based on Spark Streaming. > > I am exploring Spark Streaming for enterprise software and am cautiously > optimistic about it. I see huge potential to improve debuggability of Spark. > > ----- Original Message ----- > From: "Tim Smith" <secs...@gmail.com> > To: "spark users" <user@spark.apache.org> > Sent: Friday, September 12, 2014 10:09:53 AM > Subject: Stable spark streaming app > > Hi, > > Anyone have a stable streaming app running in "production"? Can you > share some overview of the app and setup like number of nodes, events > per second, broad stream processing workflow, config highlights etc? > > Thanks, > > Tim > > --------------------------------------------------------------------- > To unsubscribe, e-mail: user-unsubscr...@spark.apache.org For > additional commands, e-mail: user-h...@spark.apache.org > --------------------------------------------------------------------- To unsubscribe, e-mail: user-unsubscr...@spark.apache.org For additional commands, e-mail: user-h...@spark.apache.org
