Perché l'avvio di StreamingContext non riesce con "IllegalArgumentException: requisito non riuscito: non sono state registrate operazioni di output, quindi nulla da eseguire"?

Question

Sto cercando di eseguire uno Spark Streaming esempio con Twitter come fonte come segue:

public static void main (String.. args) { 

    SparkConf conf = new SparkConf().setAppName("Spark_Streaming_Twitter").setMaster("local"); 
     JavaSparkContext sc = new JavaSparkContext(conf);  
     JavaStreamingContext jssc = new JavaStreamingContext(sc, new Duration(2));  
     JavaSQLContext sqlCtx = new JavaSQLContext(sc);  


     String[] filters = new String[] {"soccer"}; 

     JavaReceiverInputDStream<Status> receiverStream = TwitterUtils.createStream(jssc,filters); 



     jssc.start(); 
     jssc.awaitTermination(); 

} 

Ma sto ottenendo la seguente eccezione

Exception in thread "main" java.lang.AssertionError: assertion failed: No output streams registered, so nothing to execute 
    at scala.Predef$.assert(Predef.scala:179) 
    at org.apache.spark.streaming.DStreamGraph.validate(DStreamGraph.scala:158) 
    at org.apache.spark.streaming.StreamingContext.validate(StreamingContext.scala:416) 
    at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:437) 
    at org.apache.spark.streaming.api.java.JavaStreamingContext.start(JavaStreamingContext.scala:501) 
    at org.learning.spark.TwitterStreamSpark.main(TwitterStreamSpark.java:53) 

Qualsiasi suggerimento come risolvere questo problema?

Answer 1

When an output operator is called, it triggers the computation of a stream.

Senza operatore di output su DStream non viene richiamato alcun calcolo. in pratica è necessario richiamare qualsiasi metodo seguito in funzione

print() 
foreachRDD(func) 
saveAsObjectFiles(prefix, [suffix]) 
saveAsTextFiles(prefix, [suffix]) 
saveAsHadoopFiles(prefix, [suffix]) 

http://spark.apache.org/docs/latest/streaming-programming-guide.html#output-operations

è possibile anche prima applicare le trasformazioni e le funzioni di uscita quindi anche, se necessario.

Answer 2

Inoltre - torto - fallisce accusare questo problema, ma la vera causa è il non più numeri tra le durate finestra scorrevole, dallo streaming di ingresso e le finestre temporali RDD. Si registra solo un avviso: a risolvere il problema, e il contesto si ferma in mancanza: D

Answer 3

Exception in thread "main" java.lang.AssertionError: assertion failed: No output streams registered, so nothing to execute

TL; DR utilizzare uno dei disponibili output operators come print, saveAsTextFiles o foreachRDD (o meno spesso usato saveAsObjectFiles o saveAsHadoopFiles).

In altre parole, è necessario utilizzare un operatore di uscita tra le seguenti righe nel codice:

JavaReceiverInputDStream<Status> receiverStream = TwitterUtils.createStream(jssc,filters); 
// --> The output operator here <-- 
jssc.start(); 

Citando Output Operations on DStreams (la mia evidenziando) della documentazione di Spark ufficiale:

Output operations allow DStream's data to be pushed out to external systems like a database or a file systems. Since the output operations actually allow the transformed data to be consumed by external systems, they trigger the actual execution of all the DStream transformations (similar to actions for RDDs).

Il punto è che senza un operatore di output hai "nessun flusso di output registrato, quindi niente da eseguire".

Come notato da un commentatore, è necessario utilizzare una trasformazione di output, ad es. print o foreachRDD, prima di avviare StreamingContext.

---

Internamente, ogni volta che si utilizza uno degli operatori di uscita disponibili, ad es. print o foreach, DStreamGraph è richiesto a add an output stream.

È possibile trovare la registrazione al successivo new ForEachDStream is created and registered (che corrisponde esattamente a add it as an output stream).