Flume + Spark - Speichern von DStream in HDFS

Question

Ich habe Kanal Stream, die ich in HDFS über Funken speichern möchte. Im Folgenden finden Sie Funken Code, den ich

object FlumePull { 
    def main(args: Array[String]) { 
    if (args.length < 2) { 
     System.err.println(
     "Usage: FlumePollingEventCount <host> <port>") 
     System.exit(1) 
    } 

    val batchInterval = Milliseconds(60000) 
    val sparkConf = new SparkConf().setAppName("FlumePollingEventCount") 
    val ssc = new StreamingContext(sparkConf, batchInterval) 
    val stream = FlumeUtils.createPollingStream(ssc, "localhost", 9999) 

    stream.map(x => x + "!!!!") 
      .saveAsTextFiles("/user/root/spark/flume_Map_", "_Mapout") 

    ssc.start() 
    ssc.awaitTermination() 
    } 
} 

leite Als ich meine spsark Streaming-Job starten, wird es ausgegeben speichert in HDFS aber Ausgang ist so etwas wie dieses:

[root@sandbox ~]# hadoop fs -cat /user/root/spark/flume_Map_-1459450380000._Mapout/part-00000 
org.apache.spark.streaming.flume.SparkFlumeEvent@1b9bd2c9!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@33fd3a48!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@35fd67a2!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@f9ed85f!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@58f4cfc!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@307373e!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@4ebbc8ff!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@a8905bb!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@29d73d64!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@71ff85b1!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@3ea261ef!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@16cbb209!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@17157890!!!! 
org.apache.spark.streaming.flume.SparkFlumeEvent@29e41c7!!!! 

Es speichert Gerinne Ereignis statt Daten aus Flume. Wie erhält man Daten daraus?

Dank

Answer 1

Sie müssen die zugrunde liegenden Puffer aus der SparkFlumeEvent und speichern extrahieren. Zum Beispiel, wenn Ihr Ereigniskörper ein String:

stream.map(x => new String(x.event.getBody.array) + "!!!!") 
     .saveAsTextFiles("/user/root/spark/flume_Map_", "_Mapout") 

ist