• Spark【Spark Streaming】

    1、基本数据源

    1.1、文件流

    在spark Shell 下运行：

    1. [lyh@hadoop102 spark-yarn-3.2.4]$ spark-shell 
    2. Setting default log level to "WARN".
    3. To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
    4. 2022-09-08 08:56:21,875 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    5. Spark context Web UI available at http://hadoop102:4040
    6. Spark context available as 'sc' (master = local[*], app id = local-1662598583370).
    7. Spark session available as 'spark'.
    8. Welcome to
    9.       ____              __
    10.      / __/__  ___ _____/ /__
    11.     _\ \/ _ \/ _ `/ __/  '_/
    12.    /___/ .__/\_,_/_/ /_/\_\   version 3.2.4
    13.       /_/
    14.          
    15. Using Scala version 2.12.15 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_241)
    16. Type in expressions to have them evaluated.
    17. Type :help for more information.
    19. scala> import org.apache.spark.streaming._
    20. import org.apache.spark.streaming._
    22. scala> val ssc = new StreamingContext(sc,Seconds(20))
    23. ssc: org.apache.spark.streaming.StreamingContext = org.apache.spark.streaming.StreamingContext@379899f4
    25. scala> val lines = ssc.textFileStream("file:///home/lyh/streaming/logfile")
    26. lines: org.apache.spark.streaming.dstream.DStream[String] = org.apache.spark.streaming.dstream.MappedDStream@531245fe
    28. scala> val kv = lines.map((_,1)).reduceByKey(_+_)
    29. kv: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.ShuffledDStream@c207c10
    31. scala> kv.print()
    33. scala> ssc.start()
    35. ------------------------------------------
    36. Time: 1662598860000 ms
    37. -------------------------------------------
    39. -------------------------------------------
    40. Time: 1662598880000 ms
    41. -------------------------------------------
    43. -------------------------------------------
    44. Time: 1662598900000 ms
    45. -------------------------------------------
    46. (c#,1)
    47. (hh,1)
    48. (h,1)
    49. (javafx,1)
    50. (spark,1)
    51. (hadoop,1)
    52. (js,1)
    53. (java,1)
    54. (s,1)
    55. (c,1)

    执行后立即新建终端在  /home/lyh/streaming/logfile 目录下创建文件并写入数据

    1.2、Socket 套接字流

    1. // todo 创建环境对象
    2.     val conf = new SparkConf()
    3.     conf.setAppName("word count").setMaster("local[*]")
    4.     val ssc = new StreamingContext(conf,Seconds(3))
    5.  
    6.     // todo 逻辑处理
    7.     // 获取端口数据(Socket)
    8.     val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    9.     val words: DStream[String] = lines.flatMap(_.split(" "))
    10.     val word: DStream[(String,Int)] = words.map((_, 1))
    11.     val wordCount: DStream[(String,Int)] = word.reduceByKey(_ + _)
    12.     wordCount.print()
    13.     // todo 关闭环境
    14.     // 由于SparkStreaming的采集器是长期运行的,所以不能直接关闭
    15.     // 而且main方法的关闭也会使SparkStreaming的采集器关闭
    16.     ssc.start()
    17.     // 等待采集器关闭
    18.     ssc.awaitTermination()

    启动 NetCat

    1. > nc -lp 9999 
    2. > hello world
    3. > hello spark
    4. > ...

    运行结果： 

    1.3、自定义 Socket 数据源

    通过自定义 Socket 实现数据源不断产生数据

    1. import java.io.PrintWriter
    2. import java.net.{ServerSocket, Socket}
    3. import scala.io.Source
    4.  
    5. /**
    6.  * 通过自定义的Socket来不断给客户端发送数据
    7.  */
    8. object MySocketReceiver {
    9.  
    10.   def index(length: Int): Int = {
    11.     val rdm = new java.util.Random()
    12.     rdm.nextInt(length)
    13.   }
    14.  
    15.   def main(args: Array[String]): Unit = {
    16.  
    17.     val fileName = "input/1.txt"
    18.     val lines: List[String] = Source.fromFile(fileName).getLines().toList
    19.  
    20.     val listener: ServerSocket = new ServerSocket(9999)
    21.  
    22.     while(true){
    23.       val socket: Socket = listener.accept()
    24.       new Thread(){
    25.         override def run(){
    26.           val out: PrintWriter = new PrintWriter(socket.getOutputStream,true)
    27.           while (true){
    28.             Thread.sleep(1000)
    29.             val content = lines(index(lines.length)) // 源源不断,每次打印list的第(1~length)随机行
    30.             println(content)
    31.             out.write(content + '\n')
    32.             out.flush()
    33.           }
    34.           socket.close()
    35.         }
    36.       }.start()
    37.     }
    38.  
    39.   }
    40. }

    定义一个处理器接收自定义数据源端口发送过来的数据。

    1. def main(args: Array[String]): Unit = {
    2.  
    3.     // todo 创建环境对象
    4.     val conf = new SparkConf()
    5.     conf.setAppName("word count").setMaster("local[*]")
    6.     val ssc = new StreamingContext(conf,Seconds(3))
    7.  
    8.     // todo 逻辑处理
    9.     // 获取端口数据(Socket)
    10.     val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    11.     val words: DStream[String] = lines.flatMap(_.split(" "))
    12.     val word: DStream[(String,Int)] = words.map((_, 1))
    13.     val wordCount: DStream[(String,Int)] = word.reduceByKey(_ + _)
    14.     wordCount.print()
    15.     // todo 关闭环境
    16.     // 由于SparkStreaming的采集器是长期运行的,所以不能直接关闭
    17.     ssc.start()
    18.     // 等待采集器关闭
    19.     ssc.awaitTermination()
    20.  
    21.   }

    先运行我们的数据源，再运行处理器：

    处理器：

    1.4、RDD 队列流

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.rdd.RDD
    3. import org.apache.spark.streaming.dstream.{DStream, InputDStream}
    4. import org.apache.spark.streaming.{Seconds, StreamingContext}
    5.  
    6. import scala.collection.mutable
    7.  
    8. object SparkStreaming02_RDDStream {
    9.  
    10.   def main(args: Array[String]): Unit = {
    11.  
    12.     // 1. 初始化配置信息
    13.     val conf = new SparkConf()
    14.     conf.setAppName("rdd Stream").setMaster("local[*]")
    15.  
    16.     // 2.初始化SparkStreamingContext
    17.     val ssc = new StreamingContext(conf,Seconds(4))
    18.  
    19.     // 3.创建RDD队列
    20.     val rddQueue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]()
    21.  
    22.     // 4.创建QueueInputStream
    23.     // oneAtATime = true 默认，一次读取队列里面的一个数据
    24.     // oneAtATime = false， 按照设定的时间，读取队列里面数据
    25.     val inputStream: InputDStream[Int] = ssc.queueStream(rddQueue,oneAtATime = false)
    26.  
    27.     // 5. 处理队列中的RDD数据
    28.     val sumStream: DStream[Int] = inputStream.reduce(_ + _)
    29.  
    30.     // 6. 打印结果
    31.     sumStream.print()
    32.  
    33.     // 7.启动任务
    34.     ssc.start()
    35.  
    36.     // 8.向队列中放入RDD
    37.     for(i <- 1 to 5){
    38.       rddQueue += ssc.sparkContext.makeRDD(1 to 5)
    39.       Thread.sleep(2000)
    40.     }
    41.  
    42.     // 9. 等待数据源进程停止后关闭
    43.     ssc.awaitTermination()
    44.   }
    45.  
    46. }

    2、高级数据源

    2.1、Kafka 数据源

    2.1.1、消费者程序处理流数据

    1. import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
    2. import org.apache.kafka.common.serialization.StringDeserializer
    3. import org.apache.spark.SparkConf
    4. import org.apache.spark.streaming.dstream.{DStream, InputDStream}
    5. import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    6. import org.apache.spark.streaming.{Seconds, StreamingContext}
    7.  
    8. object SparkStreaming03_Kafka {
    9.  
    10.   def main(args: Array[String]): Unit = {
    11.  
    12.     val conf = new SparkConf().setMaster("local[*]").setAppName("kafka source")
    13.     val ssc = new StreamingContext(conf,Seconds(3))
    14.  
    15.     // 定义Kafka参数: kafka集群地址、消费者组名称、key序列化、value序列化
    16.     val kafkaPara: Map[String,Object] = Map[String,Object](
    17.       ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
    18.       ConsumerConfig.GROUP_ID_CONFIG ->"lyh",
    19.       ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    20.       ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
    21.     )
    22.  
    23.     // 读取Kafka数据创建DStream
    24.     val kafkaDStream: InputDStream[ConsumerRecord[String,String]] = KafkaUtils.createDirectStream[String,String](
    25.       ssc,
    26.       LocationStrategies.PreferConsistent,  //优先位置
    27.       ConsumerStrategies.Subscribe[String,String](Set("testTopic"),kafkaPara) // 消费策略：（订阅多个主题，配置参数）
    28.     )
    29.  
    30.     // 将每条消息的KV取出
    31.     val valueDStream: DStream[String] = kafkaDStream.map(_.value())
    32.  
    33.     // 计算WordCount
    34.     valueDStream.flatMap(_.split(" "))
    35.       .map((_,1))
    36.       .reduceByKey(_+_)
    37.       .print()
    38.  
    39.     // 开启任务
    40.     ssc.start()
    41.     ssc.awaitTermination()
    42.  
    43.   }
    44.  
    45. }

    2.1.2、生产者生产数据

    （1）kafka 端生产数据

    启动 Kafka 集群

    创建 Topic（指定一个分区三个副本）： 

    kafka-topics.sh --bootstrap-server hadoop102:9092 --topic  --create --partitions 1 --replication-factor 3

     查看是否生成 Topic：

    kafka-topics.sh --bootstrap-server hadoop102:9092 --list

    生产者生产数据：

    1. > kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic 
    2. > hello world
    3. > hello spark
    4. > ...
    （2）编写生产者程序
    1. package com.lyh
    2.  
    3. import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
    4. import org.apache.kafka.common.serialization.StringDeserializer
    5. import org.apache.spark.SparkConf
    6. import org.apache.spark.streaming.dstream.{DStream, InputDStream}
    7. import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    8. import org.apache.spark.streaming.{Seconds, StreamingContext}
    9.  
    10. object SparkStreaming03_Kafka {
    11.  
    12.   def main(args: Array[String]): Unit = {
    13.  
    14.     val conf = new SparkConf().setMaster("local[*]").setAppName("kafka source")
    15.     val ssc = new StreamingContext(conf,Seconds(3))
    16.  
    17.     // 定义Kafka参数: kafka集群地址、消费者组名称、key序列化、value序列化
    18.     val kafkaPara: Map[String,Object] = Map[String,Object](
    19.       ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
    20.       ConsumerConfig.GROUP_ID_CONFIG ->"lyh",
    21.       ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    22.       ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
    23.     )
    24.  
    25.     // 读取Kafka数据创建DStream
    26.     val kafkaDStream: InputDStream[ConsumerRecord[String,String]] = KafkaUtils.createDirectStream[String,String](
    27.       ssc,
    28.       LocationStrategies.PreferConsistent,  //优先位置
    29.       ConsumerStrategies.Subscribe[String,String](Set("testTopic"),kafkaPara) // 消费策略：（订阅多个主题，配置参数）
    30.     )
    31.  
    32.     // 将每条消息的KV取出
    33.     val valueDStream: DStream[String] = kafkaDStream.map(_.value())
    34.  
    35.     // 计算WordCount
    36.     valueDStream.flatMap(_.split(" "))
    37.       .map((_,1))
    38.       .reduceByKey(_+_)
    39.       .print()
    40.  
    41.    // 开启任务
    42.     ssc.start()
    43.     ssc.awaitTermination()
    44.  
    45.   }
    46.  
    47. }

    3、转换操作

    3.1、无状态转换操作

    3.2、有状态转换操作

    3.1.1、滑动窗口转换操作

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.streaming.dstream.DStream
    3. import org.apache.spark.streaming.{Seconds, StreamingContext}
    4.  
    5. object SparkStreaming05_Window {
    6.   def main(args: Array[String]): Unit = {
    7.     val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming window")
    8.     val ssc = new StreamingContext(conf,Seconds(3))
    9.  
    10.     val lines:DStream[String] = ssc.socketTextStream("localhost", 9999)
    11.  
    12.     val word_kv = lines.map((_, 1))
    13.  
    14.     /**
    15.      * 收集器收集RDD合成DStream: 3s 窗口范围: 12s 窗口滑动间隔: 6s/次
    16.      * 1. windowLength：表示滑动窗口的长度，即窗口内包含的数据的时间跨度。它是一个Duration对象，用于指定窗口的时间长度。
    17.      * 2. slideInterval：表示滑动窗口的滑动间隔，即每隔多长时间将窗口向右滑动一次。同样是一个Duration对象。
    18.      * 返回一个新的 DStream
    19.      **/
    20.     val wordToOneByWindow:DStream[(String,Int)] = word_kv.window(Seconds(12), Seconds(6))
    21.  
    22.     // 窗口每滑动一次(6s),对窗口内的数据进行一次聚合操作.
    23.  
    24.     val res: DStream[(String,Int)] = wordToOneByWindow.reduceByKey(_ + _)
    25.  
    26.     res.print()
    27.  
    28.     ssc.start()
    29.     ssc.awaitTermination()
    30.   }
    31. }

    3.1.2、updateStateByKey

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.streaming.dstream.DStream
    3. import org.apache.spark.streaming.{Seconds, StreamingContext}
    4.  
    5. /**
    6.  * DStream 有状态转换操作之 updateStateByKey(func) 转换操作
    7.  */
    8. object SparkStreaming04_State {
    9.  
    10.   def main(args: Array[String]): Unit = {
    11.     val conf = new SparkConf().setMaster("local[*]").setAppName("kafka state")
    12.     val ssc = new StreamingContext(conf,Seconds(3))
    13.  
    14.     /**
    15.      * 设置检查点目录的作用是为了确保Spark Streaming应用程序的容错性和可恢复性。
    16.      * 在Spark Streaming应用程序运行过程中，它会将接收到的数据分成一批批进行处理。
    17.      * 通过设置检查点目录，Spark Streaming会定期将当前的处理状态、接收到的数据偏移量等信息保存到可靠的存储系统中，
    18.      * 比如分布式文件系统（如HDFS）或云存储服务（如Amazon S3）。
    19.      * 一旦应用程序出现故障或崩溃，它可以从最近的检查点中恢复状态，并从上次处理的位置继续处理数据，从而确保数据的完整性和一致性。
    20.      */
    21.     //检查点的路径如果是本地路径要+ file:// 否则认为是 hdfs路径 / 开头
    22.     ssc.checkpoint("file:///D://IdeaProject/SparkStudy/data/")  //设置检查点,检查点具有容错机制
    23.  
    24.     val lines: DStream[String] = ssc.socketTextStream("localhost",9999)
    25.  
    26.     val word_kv = lines.map((_, 1))
    27.  
    28.     val stateDStream: DStream[(String, Int)] = word_kv.updateStateByKey(
    29.       /** 参数是一个函数
    30.        1. Seq[Int]: 当前key对应的所有value值的集合,因为我们的value是Int,所以这里也是Int
    31.        2. Option[Int]: 当前key的历史状态,对于wordCount,历史值就是上一个DStream中这个key的value计算结果(求和结果)
    32.        Option 是 Scala 中用来表示可能存在或可能不存在的值的容器，是一种避免空引用（null reference）问题的模式。
    33.        Option[Int] 有两个可能的实例：
    34.           (1) Some(value: Int)：表示一个包含 Int 类型值的 Option。
    35.           (2) None：表示一个空的 Option，不包含任何值。
    36.       **/
    37.       (values: Seq[Int], state: Option[Int]) => {
    38.         val currentCount = values.foldLeft(0)(_ + _)
    39.         val previousCount = state.getOrElse(0)
    40.         Option(currentCount + previousCount)
    41.       }
    42.     )
    43.  
    44.     stateDStream.print()
    45.     stateDStream.saveAsTextFiles("./out") //输出结果保存到 文本文件中
    46.     ssc.start()
    47.     ssc.awaitTermination()
    48.   }
    49. }

    4、输出操作

    4.1、输出到文本文件

    上面 3.1.2 中就保存DStream输出到了本地：

    stateDStream.saveAstextFiles("./out")

    4.2、输出到MySQL数据库

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.rdd.RDD
    3. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    4. import org.apache.spark.streaming.{Seconds, StreamingContext}
    5.  
    6. import java.sql.{Connection, PreparedStatement}
    7.  
    8. object NetWorkWordCountStateMySQL {
    9.  
    10.   def main(args: Array[String]): Unit = {
    11.  
    12.     val updateFunc = (values: Seq[Int],state: Option[Int]) => {
    13.       val currentCount = values.foldLeft(0)(_+_)
    14.       val previousCount = state.getOrElse(0)
    15.       Some(currentCount + previousCount)
    16.     }
    17.  
    18.     val conf = new SparkConf().setMaster("local[*]").setAppName("state mysql")
    19.     val ssc = new StreamingContext(conf,Seconds(5))
    20.     // file:\\ 代表本地文件系统 如果用的是 /user/... 这种形式是 HDFS 文件系统 需要启动Hadoop集群
    21.     ssc.checkpoint("file:\\D:\\IdeaProjects\\SparkStudy\\data\\state")
    22.  
    23.     val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    24.     val word_kv: DStream[(String, Int)] = lines.flatMap(_.split(" ").map((_, 1))).reduceByKey(_ + _)
    25.  
    26.     val stateDStream: DStream[(String, Int)] = word_kv.updateStateByKey[Int](updateFunc)
    27.     stateDStream.print()
    28.  
    29.     stateDStream.foreachRDD( rdd=> {
    30.  
    31.       def func(records: Iterator[(String,Int)]): Unit ={
    32.         var conn: Connection = null
    33.         var stmt: PreparedStatement = null
    34.         try{
    35.           conn = DBUtils.getConnection("jdbc:mysql://127.0.0.1:3306/spark","root","Yan1029.")
    36.           records.foreach(p=>{
    37.             val sql = "insert into wordcount values (?,?)"
    38.             stmt = conn.prepareStatement(sql)
    39.             stmt.setString(1,p._1.trim)
    40.             stmt.setInt(2,p._2)
    41.             stmt.executeUpdate()    //不executeUpdate就不会写入数据库
    42.           })
    43.         }catch {
    44.           case e: Exception => e.printStackTrace()
    45.         }finally {
    46. //          if (stmt!=null) stmt.close()
    47. //          DBUtils.close()
    48.         }
    49.       }
    50.       val repartitionedRDD: RDD[(String,Int)] = rdd.repartition(3)  //扩大分区用 repartition
    51.       repartitionedRDD.foreachPartition(func)
    52.     })
    53.     ssc.start()
    54.     ssc.awaitTermination()
    55.   }
    56.  
    57. }

    运行结果：

    5、优雅的关闭和恢复数据

    5.1、关闭SparkStreaming

            流式任务通常都需要7*24小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所以配置优雅的关闭就显得至关重要了。
           关闭方式：我们通常使用外部文件系统来控制内部程序关闭。

    1. package com.lyh
    2.  
    3. import org.apache.hadoop.conf.Configuration
    4. import org.apache.hadoop.fs.{FileSystem, Path}
    5. import org.apache.spark.SparkConf
    6. import org.apache.spark.streaming.dstream.DStream
    7. import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
    8.  
    9. import java.net.URI
    10.  
    11. object SparkStreaming06_Close {
    12.   def main(args: Array[String]): Unit = {
    13.     val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming close")
    14.     val ssc = new StreamingContext(conf,Seconds(3))
    15.  
    16.     val lines:DStream[String] = ssc.socketTextStream("localhost", 9999)
    17.     val word_kv = lines.map((_, 1))
    18.  
    19.     word_kv.print()
    20.  
    21.     ssc.start()
    22.  
    23.     // 再创建一个线程去关闭
    24.     new Thread(new MonitorStop(ssc)).start()
    25.  
    26.     ssc.awaitTermination()  //阻塞当前main线程
    27.   }
    28. }
    29.  
    30. class MonitorStop(ssc: StreamingContext) extends Runnable{
    31.   override def run(): Unit = {
    32.     while (true){ // 一直轮询判断
    33.       Thread.sleep(5000)  //每5s检查一遍
    34.       val fs: FileSystem = FileSystem.get(new URI("hdfs://hadoop102:9000"),new Configuration(),"lyh")
    35.       val exists: Boolean = fs.exists(new Path("hdfs://hadoop102:9000/stopSpark"))
    36.       if (exists) { //如果比如（MySQL出现了一行数据、Zookeeper的某个节点出现变化、hdfs是否存在某个目录...）就关闭
    37.         val state: StreamingContextState = ssc.getState()
    38.         if (state == StreamingContextState.ACTIVE){
    39.           // 优雅地关闭-处理完当前的数据再关闭
    40.           // 计算节点不再接受新的数据，而是把现有的数据处理完毕，然后关闭
    41.           ssc.stop(true,true)
    42.           System.exit(0)
    43.         }
    44.       }
    45.     }
    46.   }
    47. }

    5.2、恢复检查点的数据

    使用 getActiveOrCreate 的方法来对上一个失败的 Spark 任务进行数据恢复（通过检查点来进行恢复）

    方法说明：

            若Application为首次重启，将创建一个新的StreamingContext实例；如果Application从失败中重启，从checkpoint目录导入checkpoint数据来重新创建StreamingContext实例。

    1. import org.apache.spark.SparkConf
    2. import org.apache.spark.streaming.dstream.DStream
    3. import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
    4.  
    5. import java.net.URI
    6.  
    7. object SparkStreaming07_Resume {
    8.   def main(args: Array[String]): Unit = {
    9.  
    10.     //好处：若Application为首次重启，将创建一个新的StreamingContext实例；如果Application从失败中重启，从checkpoint目录导入checkpoint数据来重新创建StreamingContext实例。
    11.     val ssc: StreamingContext = StreamingContext.getActiveOrCreate("file:\\D:\\IdeaProjects\\SparkStudy\\data\\state", () => {
    12.       val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming resume")
    13.       val ssc = new StreamingContext(conf, Seconds(3))
    14.  
    15.       val lines: DStream[String] = ssc.socketTextStream("localhost", 9999)
    16.       val word_kv = lines.map((_, 1))
    17.  
    18.       word_kv.print()
    19.  
    20.       ssc
    21.     })
    22.     // 依然设置检查点 防止application失败后丢失数据
    23.     ssc.checkpoint("file:\\D:\\IdeaProjects\\SparkStudy\\data\\state")
    24.  
    25.     ssc.start()
    26.     ssc.awaitTermination()  //阻塞当前main线程
    27.   }
    28. }

  • 相关阅读:
    你写过的最蠢的代码是？——前端篇
    牛客网：迷宫问题
    Vue3引入腾讯地图，点击坐标后实时获取经纬度
    Angular 学习 之 Hello World !
    Docker实战技巧（一）：Kubernetes基础操作实战
    自然语言处理 | 语言模型(LM) 浅析
    CSS 笔记（十三）：常用单位 & 适配方案（移动端）
    DOM对非表单元素与表单元素的属性操作
    React@16.x（25）useReducer
    推荐系统-指标：ctr、cvr
  • 原文地址：https://blog.csdn.net/m0_64261982/article/details/133964098