1、数据源：TCP Socket从哪里读取实时数据，然后进行实时分析
2、数据终端：输出控制台结果数据输出到哪里
3、功能：对每批次数据实时统计，时间间隔BatchInterval：1s

# 官方入门案例运行：词频统计
/export/server/spark/bin/run-example --master local[2] streaming.NetworkWordCount node1.oldlu.cn 9999

1、Define the input sources by creating input DStreams.定义从哪个数据源接收流式数据，封装到DStream中
2、Define the streaming computations by applying transformation and output operations to DStreams.针对业务调用DStream中函数，进行数据处理和输出
3、Start receiving data and processing it using streamingContext.start().
4 、 Wait for the processing to be stopped (manually or due to any error) usingstreamingContext.awaitTermination().
5、The processing can be manually stopped using streamingContext.stop().启动流式应用，并且一直等待程序终止（人为或异常），最后停止运行

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
 * 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
 */
object StreamingWordCount {
  def main(args: Array[String]): Unit = {
    // TODO: 1. 构建StreamingContext流式上下文实例对象
    val ssc: StreamingContext = {
      // a. 创建SparkConf对象，设置应用配置信息
      val sparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[3]")
      // b.创建流式上下文对象, 传递SparkConf对象，TODO: 时间间隔 -> 用于划分流式数据为很多批次Batch
      val context = new StreamingContext(sparkConf, Seconds(5))
      // c. 返回
      context
    }
    // TODO: 2. 从数据源端读取数据，此处是TCP Socket读取数据
    /*
    def socketTextStream(
    hostname: String,
    port: Int,
    storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
    ): ReceiverInputDStream[String]
    */
    val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream(
      "node1.oldlu.cn", 9999
    )
    // TODO: 3. 对每批次的数据进行词频统计
    val resultDStream: DStream[(String, Int)] = inputDStream
      // 过滤不合格的数据
      .filter(line => null != line && line.trim.length > 0)
      // 按照分隔符划分单词
      .flatMap(line => line.trim.split("\\s+"))
      // 转换数据为二元组，表示每个单词出现一次
      .map(word => (word, 1))
      // 按照单词分组，聚合统计
      .reduceByKey((tmp, item) => tmp + item)
    // TODO: 4. 将结果数据输出 -> 将每批次的数据处理以后输出
    resultDStream.print(10)
    // TODO: 5. 对于流式应用来说，需要启动应用
    ssc.start()
    // 流式应用启动以后，正常情况一直运行（接收数据、处理数据和输出数据），除非人为终止程序或者程序异常停止
    ssc.awaitTermination()
    // 关闭流式应用(参数一：是否关闭SparkContext，参数二：是否优雅的关闭）
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
 * 从TCP Socket 中读取数据，对每批次（时间为5秒）数据进行词频统计，将统计结果输出到控制台。
 * TODO: 从多个Socket读取流式数据，进行union合并
 */
object StreamingDStreamUnion {
  def main(args: Array[String]): Unit = {
    // TODO: 1. 构建StreamingContext流式上下文实例对象
    val ssc: StreamingContext = {
      // a. 创建SparkConf对象，设置应用配置信息
      val sparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[4]")
      // b.创建流式上下文对象, 传递SparkConf对象，TODO: 时间间隔 -> 用于划分流式数据为很多批次Batch
      val context = new StreamingContext(sparkConf, Seconds(5))
      // c. 返回
      context
    }
    // TODO: 2. 从数据源端读取数据，此处是TCP Socket读取数据
    /*
    def socketTextStream(
    hostname: String,
    port: Int,
    storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
    ): ReceiverInputDStream[String]
    */
    val inputDStream01: DStream[String] = ssc.socketTextStream("node1.oldlu.cn", 9999)
    val inputDStream02: DStream[String] = ssc.socketTextStream("node1.oldlu.cn", 9988)
    // 合并两个DStream流
    val inputDStream: DStream[String] = inputDStream01.union(inputDStream02)
    // TODO: 3. 对每批次的数据进行词频统计
    val resultDStream: DStream[(String, Int)] = inputDStream
      // 过滤不合格的数据
      .filter(line => null != line && line.trim.length > 0)
      // 按照分隔符划分单词
      .flatMap(line => line.trim.split("\\s+"))
      // 转换数据为二元组，表示每个单词出现一次
      .map(word => (word, 1))
      // 按照单词分组，聚合统计
      .reduceByKey((tmp, item) => tmp + item)
    // TODO: 4. 将结果数据输出 -> 将每批次的数据处理以后输出
    resultDStream.print(10)
    // TODO: 5. 对于流式应用来说，需要启动应用
    ssc.start()
    // 流式应用启动以后，正常情况一直运行（接收数据、处理数据和输出数据），除非人为终止程序或者程序异常停止
    ssc.awaitTermination()
    // 关闭流式应用(参数一：是否关闭SparkContext，参数二：是否优雅的关闭）
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

// TODO: 2. 从数据源端读取数据，此处是TCP Socket读取数据
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream(
  "node1.oldlu.cn", //
  9999, //
  // TODO: 设置Block存储级别为先内存，不足磁盘，副本为1
  storageLevel = StorageLevel.MEMORY_AND_DISK
)

官方案例：
BatchInterval： 1s = 1000ms = 5 * BlockInterval
每批次RDD数据中，有5个Block，每个Block就是RDD一个分区数据

// TODO: 3. 对每批次的数据进行词频统计
val resultDStream: DStream[(String, Int)] = inputDStream
  // 过滤不合格的数据
  .filter(line => null != line && line.trim.length > 0)
  // 按照分隔符划分单词
  .flatMap(line => line.trim.split("\\s+"))
  // 转换数据为二元组，表示每个单词出现一次
  .map(word => (word, 1))
  // 按照单词分组，聚合统计
  .reduceByKey((tmp, item) => tmp + item)
// TODO: 4. 将结果数据输出 -> 将每批次的数据处理以后输出
resultDStream.print(10)