Spark Streaming 快速入门系列(2) | DStream 入门WorldCount案例与解析

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>2.1.1</version>
</dependency>

package com.buwenbuhuo.spark.streaming.day01

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author 不温卜火
 * @create 2020-08-07 11:41
    MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 */
object WordCount {
  def main(args: Array[String]): Unit = {
      // 1. 创建StreamingContext
      val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("WordCount")
      val ssc: StreamingContext = new StreamingContext(conf,Seconds(3))

      // 2. 从数据源创建一个流： socket，rdd队列，自定义接收器 ，kafka(重点)
      val sourceStream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop002",9999)

      // 3. 对流做各种转换
      val result: DStream[(String, Int)] = sourceStream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _)

      // 4. 行动算子 print foreach foreachRDD
      result.print()  //  把结果打印在控制台

      // 5. 启动流
      ssc.start()

     // 6. 阻止主线程退出(阻塞主线程)
    ssc.awaitTermination()



  }
}