在开篇之前需要注意的是Spark Streaming Programming Guide | NoteSpark Streaming is the previous generation of Spark’s streaming engine. There are no longer updates to Spark Streaming and it’s a legacy project. There is a newer and easier to use streaming engine in Spark called Structured Streaming. You should use Spark Structured Streaming for your streaming applications and pipelines.sparkStreaming目前已不在更新Structured Streaming 作为一个新的、更易用的流引擎用来处理我们的流任务。所以本篇的主要作用是对spark streaming的一个理解。大家可轻松阅读。文章目录一. SparkStreaming概述1. 初认识2. spark Streaming内部处理机制消费者生产者模型二. hello SparkSteaming从telnet计算wordcount1. 需求2. 实现三. sparkStreaming数据源1. File Streams ing2. 自定义数据源3. kafka数据源需求实现四. Transformations on DStreams1. 常用的算子2. UpdateStateByKey Operation有状态跨批次数据累加3. window函数4. join ing五. Output Operations on DStreams一. SparkStreaming概述1. 初认识sparkStreaming是spark 核心api的一个延伸用于流式数据的处理。数据源可以是KafkaFlume文件简单的TCP套接字等可以使用map、reduce、join、window等高阶函数去处理数据最后数据可以被推送到文件系统HDFS等、数据库甚至是实时看板。2. spark Streaming内部处理机制消费者生产者模型如下图spark Streaming的内部处理机制是接收实时流数据并按照一定的时间间隔拆分成一批批数据通过Spark Engine处理这些数据最后得到这些数据。生产者与消费者模型批数据对应的是一个RDD实例所以流数据的Dstreamcontinuous stream of data可以看成是一组RDDs一个RDD序列。通俗一点讲流数据分成一批批后通过一个先进先出的队列然后Spark Engine 从该队列中依次取出然后将批数据封装成一个RDD然后进行处理这是一个典型的生产者消费者模型。所以spark的流处理实际是将流数据封装为一个个RDD然后依然通过spark Engine去处理这些数据所以本质上Spark Streaming的计算还是由spark Core的计算引擎来实现的。DStream是离散流代表了持续不断的数据流通过上面的分析可以知道DStream中一批RDD。二. hello SparkSteaming从telnet计算wordcount1. 需求使用netcat工具向9999端口不断的发送数据通过SparkStreaming读取端口数据并每三秒统计不同单词出现的次数2. 实现依赖dependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming_2.11/artifactIdversion2.1.1/version/dependency编程启动与结果展示nc-lk9999hello atguigu三. sparkStreaming数据源在上节我们已经了解到了使用ssc.socketTextStream处理来自TCP socket的数据。接下来我们完整的看一下spark streaming支持的数据源。1. File Streams ing支持的文件系统sparkStreaming能够读取所有HDFS API兼容的文件系统文件读取a DStream can be created as viaStreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass].监控Spark Streaming 将会监控 dataDirectory 目录并不断处理移动进来的文件目前不支持嵌套目录。注意文件夹中的文件对应的数据都是相同格式的监控到的方式移动文件或重命名不可修改一旦进入目录则不允许修改修改也不会读取ing2. 自定义数据源packagecom.aura.bigdata.SparkStreamingimportjava.io.{BufferedReader,InputStream,InputStreamReader}importjava.net.Socketimportorg.apache.spark.SparkConfimportorg.apache.spark.storage.StorageLevelimportorg.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}importorg.apache.spark.streaming.receiver.Receiverimportorg.apache.spark.streaming.{Seconds,StreamingContext}//wordcountobjectSparkStreaming03_ReadfromReceiver{defmain(args:Array[String]):Unit{valsparkConfnewSparkConf().setMaster(local[*]).setAppName(11)valstreamingContextnewStreamingContext(sparkConf,Seconds(3))valrDS:ReceiverInputDStream[String]streamingContext.receiverStream(newMyReceiver(hadoop03,9999))valwordDStream:DStream[String]rDS.flatMap(lineline.split( ))valmapDStream:DStream[(String,Int)]wordDStream.map((_,1))valwordToSumDStream:DStream[(String,Int)]mapDStream.reduceByKey(__)wordToSumDStream.print()//因为数据是不断发送的所以streamingContext采集程序不能停止//streamingContext.stop()streamingContext.start()streamingContext.awaitTermination()}}//自定义socket数据流规则当输入END字符时socket关闭连接。//1继承ReceiverclassMyReceiver(host:String,port:Int)extendsReceiver[String](StorageLevel.MEMORY_ONLY){varsocket:Socketnull;defreceive():Unit{//创建sokect进行网络通讯(从虚拟机上)socketnewSocket(host,port)valreadernewBufferedReader(newInputStreamReader(socket.getInputStream,utf-8))varline:Stringnullwhile((linereader.readLine())!null){//将采集到的数据//1、以“END”作为停止采集的标志if(END.equals(line)){return}else{//2、储存数据this.store(line)}}}//启动线程持续实现数据的接收overridedefonStart():Unit{newThread(newRunnable{overridedefrun():Unit{receive()}}).start()}overridedefonStop():Unit{if(socket!null){socket.close()socketnull}}}3. kafka数据源需求通过SparkStreaming从Kafka读取数据并将读取过来的数据做简单计算(WordCount)最终打印到控制台。实现依赖dependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming-kafka-0-8_2.11/artifactIdversion2.1.1/version/dependencydependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion0.11.0.2/version/dependency程序四. Transformations on DStreams1. 常用的算子DStreams support many of the transformations available on normal Spark RDD’s.算子解释map(func)一个输入一个输出flatMap(func)一个输入多个输出filter(func)根据func保留返回为ture的数据repartition(numPartitions)重新生成指定分区数的DStreamunion(otherStream)和另外一个DStream组成一个新的DStreamcount()记录元素数量reduce(func)聚合countByValue()计算source中每个key出现的频率reduceByKey(func, [numTasks])根据func规则去聚合source数据join(otherStream, [numTasks])When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) pairs with all pairs of elements for each key.cogroup(otherStream, [numTasks])When called on a DStream of (K, V) and (K, W) pairs, return a new DStream of (K, Seq[V], Seq[W]) tuples.updateStateByKey(func)返回一个新的“状态”DStream其中每个键的状态通过在键的前一个状态和键的新值上通过func规则来更新。这可用于维护每个键的任意状态数据。2. UpdateStateByKey Operation有状态跨批次数据累加功能此函数用于实现跨批次结果的累加它记录了历史状态并提供了对状态变量的访问。使用状态的更新需要做两步• 定义状态可以是一个任意的数据类型• 定义状态更新函数状态值与输入流的更新操作需要配置检查点目录会使用检查点来保存状态updateFunc方法逻辑• Seq[int]map(_,1)中的1去形成序列1,1,1,1…• Option[S]代表的是缓冲区的数据可能为null• 更新函数seq缓冲区的数据做交互返回一个新的Option更新到缓冲区最新的聚合数据3. window函数一个窗口内会处理更多批次的结果窗口需要两个参数窗口时长滑动步长注意两个参数都是批次间隔的整数倍。4. join ingJoin Operations五. Output Operations on DStreamsOutput Operations on DStreams
