使用Streaming读取集群端数据实现WordCount功能

使用Streaming读取集群端数据实现WordCount功能

1. 准备工作：注意spark和Scala版本兼容问题

2. 具体代码

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


object StrWC {
  def main(args: Array[String]): Unit = {
    //TODO 创建环境对象 -> StreamingContext创建时，需要传递两个参数
    //1. 环境配置 注意：核数必须 >接收器 因为要多出一个核来进行计算
    val sparkConf: SparkConf = new SparkConf().setAppName("StrWC").setMaster("local[*]")
    //2. 批处理的采集周期 Seconds(3)
    val ssc:StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //TODO 逻辑处理
    //1. 获取端口数据 开启集群打开端口 -> nc -lk 9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", 9999)
    lines.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()
    
    //TODO 启动采集器
    //1. 启动采集器
    ssc.start()
    //2. 等待采集器的关闭 保证应用程序一直运行 被称为长程序
    ssc.awaitTermination()
  }
}

3. 打开集群和端口，运行程序并输入