spark-streaming-flume

code/spark/spark-streaming-flume/src/main/scala/com/heibaiying/flume/PullBasedWordCount.scala

@@ -18,9 +18,8 @@ object PullBasedWordCount {
     // 1.获取输入流
     val flumeStream = FlumeUtils.createPollingStream(ssc, "hadoop001", 8888)
 
-    // 2.词频统计
+    // 2.打印输入流中的数据
-    flumeStream.map(line => new String(line.event.getBody.array()).trim)
+    flumeStream.map(line => new String(line.event.getBody.array()).trim).print()
-      .flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()
 
     ssc.start()
     ssc.awaitTermination()

code/spark/spark-streaming-flume/src/main/scala/com/heibaiying/flume/PushBasedWordCount.scala

@@ -19,9 +19,8 @@ object PushBasedWordCount {
     // 1.获取输入流
     val flumeStream = FlumeUtils.createStream(ssc, "hadoop001", 8888)
 
-    // 2.词频统计
+    // 2.打印输入流的数据
-    flumeStream.map(line => new String(line.event.getBody.array()).trim)
+    flumeStream.map(line => new String(line.event.getBody.array()).trim).print()
-      .flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()
 
     ssc.start()
     ssc.awaitTermination()