Zählen von Aufzeichnungen meiner RDDs in einem großen Dstream

Question

Ich versuche, mit einer großen RDD zu arbeiten, wie von einer Datei DStream gelesen.

Der Code sieht wie folgt aus:

val creatingFunc = {() => 
    val conf = new SparkConf() 
       .setMaster("local[10]") 
       .setAppName("FileStreaming") 
       .set("spark.streaming.fileStream.minRememberDuration", "2000000h") 
       .registerKryoClasses(Array(classOf[org.apache.hadoop.io.LongWritable], 
classOf[org.apache.hadoop.io.Text], classOf[GGSN])) 

    val sc = new SparkContext(conf) 

    // Create a StreamingContext 
    val ssc = new StreamingContext(sc, Seconds(batchIntervalSeconds)) 

    val appFile = httpFileLines 
        .map(x=> (x._1,x._2.toString())) 
        .filter(!_._2.contains("ggsnIPAddress")) 
        .map(x=>(x._1,x._2.split(","))) 

    var count=0 

    appFile.foreachRDD(s => { 
    // s.collect() throw exception due to insufficient amount of emery 
    //s.count() throw exception due to insufficient amount of memory 
    s.foreach(x => count = count + 1) 
    }) 

    println(count) 
    newContextCreated = true 
    ssc 
} 

, was ich versuche zu tun, die Zählung meines RDD..however zu bekommen, da es large..it ist wirft exception..so muss ich tun stattdessen ein foreach zu vermeiden, Daten in den Speicher zu sammeln ..

ich möchte die Zählung dann als Weg in meinem Code zu bekommen, aber es gibt immer 0 ..

gibt es eine Möglichkeit, dies zu tun?

Answer 1

Es ist nicht erforderlich, foreachRDD und rufen Sie count. Sie können die count Methode auf DStream definiert verwenden:

val appFile = httpFileLines 
       .map(x => (x._1, x._2.toString())) 
       .filter(!_._2.contains("ggsnIPAddress")) 
       .map(x => (x._1, x._2.split(","))) 

val count = appFile.count() 

Wenn das immer noch eine unzureichende Menge an Speicher-Ausnahme ergibt, müssen Sie entweder jedes Mal die Berechnung werden kleinere Chargen von Daten, oder vergrößern Sie Arbeiter Knoten die Last zu bewältigen .