Zählen mit Spark Streaming

Question

Ich habe ein Programm in Spark Streaming, das eingehende Dateien in HDFS erkennt, was ich tun möchte, ist jede Datei analysieren und testen das Vorhandensein von zwei Wörtern in jeder Datei, und an jedem Punkt wissen, wie viele Dateien enthalten diese beiden Wörter. Was ich versucht zu tun:

val recherche1 = lines.map(x => (x.split(":")(0),x.split(":")(1))).filter(x => x._2 == "mot1") 
    val recherche2 = lines.map(x => (x.split(":")(0),x.split(":")(1))).filter(x => x._2 == "mot2") 
    val n1 = recherche1.count() 
    val n2 = recherche1.count() 
    val p = n1.foreachRDD(rdd => {cont1 = rdd.count() 
    if (cont1 > 0) 
    {n2.foreachRDD(rdd => {cont2 = rdd.count() 
     if (cont2 > 0) 
     {number = number + 1} 
     else 
     {number = number} 
    })} 
    }) 

, was ich will, ist der Wert der Variablen „Zahl“ zu verwenden, so da ich es nicht in Spark-Streaming drucken können, versuche ich es in HBase zu setzen, leider es nicht funktioniert und wenn ich das Skript in funken einreichen starten, gibt es den Fehler:

adding new inputs transformations and output operations after starting a context is not supported 

weiß jemand, was ich falsch mache oder kann mir sagen, wie dies zu tun?

Vielen Dank im Voraus

Answer 1

ich die ganze Struktur des in der Frage nach der Antwort auf Ihre Frage angezeigten Code geändert: filter the lines by two words Spark Streaming und es hat funktioniert, vielleicht diese Struktur für Spark ist nicht geeignet Streaming ..