Spark Streaming编程指南【快速入门】

﹏ヽ暗。殇╰゛Y 2023-02-16 05:46 16阅读 0赞

## 一.概述 ##

Spark Streaming是核心Spark API的扩展，可实现实时数据流【**微批处理**】的可伸缩，高吞吐量，容错流处理。数据可以从像kafka，flume，kinesis或TCP套接字许多来源摄入，并且可以使用高级函数例如map，reduce，join和window来表达复杂的算法。最后，可以将处理后的数据推送到文件系统，数据库和实时仪表板。实际上，可以在数据流上应用Spark的**机器学习**和 **图计算**算法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
在内部，Spark Streaming接收实时输入数据流，并将数据分成批处理，然后由Spark引擎进行处理，以生成批处理的最终结果流。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]  
Spark Streaming提供了称为离散流或DStream的高级抽象，它表示连续的数据流。可以根据来自Kafka，Flume和Kinesis等来源的输入数据流来创建DStream，也可以通过对其他DStream应用高级操作来创建DStream。**在内部，DStream表示为RDD序列** 。

## 二.案例实战 ##

首先，我们将Spark Streaming类的名称以及从StreamingContext进行的一些隐式转换导入到我们的环境中，以便向我们需要的其他类（如DStream）添加有用的方法。StreamingContext是所有流功能的主要入口点。我们创建具有两个执行线程和5秒批处理间隔的本地StreamingContext。

val spark = SparkSession.builder().appName("StreamingSocket").master("local[2]").getOrCreate()
        val sc = spark.sparkContext
        /** * 5s计算一次 */
        val ssc = new StreamingContext(sc, Seconds(5))

使用此上下文，我们可以创建一个DStream，该DStream表示来自TCP源的流数据，指定为主机名和端口。此socketDStream表示将从数据服务器接收的数据流。此DStream中的每个记录都是一行文本。

val socketDStream = ssc.socketTextStream("master", 9999)

flatMap是一对多DStream操作，它通过从源DStream中的每个记录生成多个新记录来创建新的DStream。在这种情况下，每行将被拆分为多个单词，单词流被表示为 wordsDStream。将wordsDStream进一步映射（一对一转换）到(word, 1)成对的DStream，然后使用reduceByKey获取每批数据中单词的频率。最后，reduce.print()将打印5秒内产生的一些计数。

val words = socketDStream.flatMap(_.split(" ")).map(item => (item, 1))
        val reduce = words.reduceByKey(_ + _)
        reduce.print()

请注意，执行这些行时，Spark Streaming仅设置启动时将执行的计算，并且尚未开始任何实际处理【**懒加载**】。在完成所有转换后，要开始处理，我们最终调用ssc.start()。

ssc.start()             // Start the computation
    	ssc.awaitTermination()  // Wait for the computation to terminate

完整代码如下：

package spark2.streaming
    
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    /** * 实时统计 * Created by Administrator on 2019/10/27. */
    object StreamingSocketWordCount { 
      Logger.getLogger("org").setLevel(Level.WARN)
      def main(args: Array[String]) { 
        val spark = SparkSession.builder().appName("StreamingSocket").master("local[2]").getOrCreate()
        val sc = spark.sparkContext
        /** * 5s计算一次 */
        val ssc = new StreamingContext(sc, Seconds(5))
    
        /** * 加载socket数据 */
        val socketDStream = ssc.socketTextStream("master", 9999)
    
        val words = socketDStream.flatMap(_.split(" ")).map(item => (item, 1))
        val reduce = words.reduceByKey(_ + _)
        reduce.print()
    
        /** * 开启流计算 */
        ssc.start()
        ssc.awaitTermination()
      }
    }

启动socket作为数据服务器运行：  
![在这里插入图片描述][20200608150942557.png]  
执行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200608145102151.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200608145147556.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[20200608150942557.png]: https://img-blog.csdnimg.cn/20200608150942557.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200608151332700.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70