You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by sivarani <wh...@gmail.com> on 2014/11/04 16:01:09 UTC
Spark Streaming getOrCreate
Hi All
I am using SparkStreaming..
public class SparkStreaming{
SparkConf sparkConf = new SparkConf().setAppName("Sales");
JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new
Duration(5000));
String chkPntDir = ""; //get checkpoint dir
jssc.checkpoint(chkPntDir);
JavaSpark jSpark = new JavaSpark(); //this is where i have the business
logic
JavaStreamingContext newJSC = jSpark.callTest(jssc);
newJSC.start();
newJSC.awaitTermination();
}
where
public class JavaSpark implements Serializable{
public JavaStreamingContext callTest(JavaStreamingContext){
logic goes here
}
}
is working fine
But i try getOrCreate as i want spark streaming to run 24/7
JavaStreamingContextFactory contextFactory = new
JavaStreamingContextFactory() {
@Override
public JavaStreamingContext create() {
SparkConf sparkConf = new SparkConf().setAppName(appName).setMaster(master);
JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new
Duration(5000));
jssc.checkpoint("checkpointDir");
JavaSpark js = new JavaSpark();
JavaStreamingContext newJssc = js.callTest(jssc);// This is where all the
logic is
return newJssc;
}
JavaStreamingContext context =
JavaStreamingContext.getOrCreate(checkPointDir, contextFactory);
context.start();
context.awaitTermination();
Not working
14/11/04 19:40:37 ERROR SparkDeploySchedulerBackend: Application has been
killed. Reason: All masters are unresponsive! Giving up.
Exception in thread "Thread-37" org.apache.spark.SparkException: Job aborted
due to stage failure: All masters are unresponsive! Giving up.
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1049)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1033)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1031)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at
org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1031)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at scala.Option.foreach(Option.scala:236)
at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:635)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1234)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
14/11/04 19:40:37 ERROR JobScheduler: Error running job streaming job
1415110180000 ms.0
org.apache.spark.SparkException: Job aborted due to stage failure: All
masters are unresponsive! Giving up.
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1049)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1033)
Please help me out.
Earlier the biz logic was inside the ContextFactory but i got
org.apache.spark.SparkException: Job aborted due to stage failure: Task not
serializable: java.io.NotSerializableException:
com.zoho.zbi.spark.PaymentStreaming$1
Then i added private static final long serialVersionUID =
-5751968749110204082L; in all the method dint work either
Got
14/11/04 19:40:37 ERROR SparkDeploySchedulerBackend: Application has been
killed. Reason: All masters are unresponsive! Giving up.
Exception in thread "Thread-37" org.apache.spark.SparkException: Job aborted
due to stage failure: All masters are unresponsive! Giving up.
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1049)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1033)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1031)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at
org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1031)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at scala.Option.foreach(Option.scala:236)
at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:635)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1234)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
14/11/04 19:40:37 ERROR JobScheduler: Error running job streaming job
1415110180000 ms.0
org.apache.spark.SparkException: Job aborted due to stage failure: All
masters are unresponsive! Giving up.
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1049)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1033)
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Streaming-getOrCreate-tp18060.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Spark Streaming getOrCreate
Posted by sivarani <wh...@gmail.com>.
Anybody any luck? I am also trying to set NONE to delete key from state, will
null help? how to use scala none in java
My code goes this way
public static class ScalaLang {
public static <T> Option<T> none() {
return (Option<T>) None$.MODULE$;
}
}
Function2<List<Double>, Optional<Double>, Optional<Double>>
updateFunction =
new Function2<List<Double>, Optional<Double>,
Optional<Double>>() {
@Override public Optional<Double> call(List<Double> values,
Optional<Double> state) {
Double newSum = state.or(0D);
if(values.isEmpty()){
System.out.println("empty value");
return null; I WANT TO RETURN NONE TO DELETE KEY but when i
set ScalaLang.<>none(); it shows error
}else{
for (double i : values) {
newSum += i;
}
return Optional.of(newSum);
}
}
};
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Streaming-getOrCreate-tp18060p18139.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Spark Streaming getOrCreate
Posted by Yana <ya...@gmail.com>.
Siarani, does your spark-master look like it's still up (i.e. if you check
the UI?).
I cannot tell if you see this error on "get" or "initial create". You can
start debugging by dumping out the value of master in setMaster(master) --
especially if this failure is from the intial startup
>From the error it appears that your application is unable to (re?)connect to
the master upon checkpoint restart -- I wonder if the reason it needs to
restart is that the master went down...
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Streaming-getOrCreate-tp18060p18194.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org