大きなDstreamで自分のRDDのレコードを集計する

Question

大きなRDDをファイルDStreamとして読み込もうとしています。次のように

コードが見えます：それはlarge..itは私が行う必要がありexception..so投げているので、私がやろうとしています何

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 
} 

は私RDD..howeverのカウントを取得することですメモリにデータを収集しないようにする代わりに、foreachの..

..私はその後、私のコード内の方法として、カウントを取得したいが、それは常に0を与えるが、これを行うためにそこに方法は何ですか？

Answer 1

foreachRDDに電話する必要はありません。電話番号countです。それはまだメモリ例外の不十分な量が得られた場合のいずれかが毎回データの小さなバッチを計算する、またはあなたの負荷を処理するワーカーノードを拡大する必要があり、

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() 

：あなたはDStreamに定義されたcountメソッドを使用することができます。