精华内容
下载资源
问答
  • spark streaming

    2018-03-14 15:21:35
    spark streaming streaming
  • Spark Streaming

    2019-01-21 07:10:46
    Spark streaming tutorial, with advanced analysis.!
  • sparkstreaming

    2018-04-28 14:25:12
    本文详细的介绍了sparkstreaming的基本概念,如何使用等问题
  • SparkStreaming

    2019-04-14 13:16:06
    SparkStreaming 概述 Spark Streaming可以轻松构建可扩展的容错流应用程序 1)便于使用 2) 容错 3)Spark集成 计算框架 log-》flume-》kafka-》sparkStreaming-》hdfs || mysql -》hive||hbase Dstream 数据...

    SparkStreaming

    概述
    Spark Streaming可以轻松构建可扩展的容错流应用程序

    1)便于使用
    2) 容错
    3)Spark集成

    计算框架

    log-》flume-》kafka-》sparkStreaming-》hdfs || mysql -》hive||hbase
    在这里插入图片描述

    Dstream

    数据源来源于分布式多台机器,多台计算
    Dframe 属于离散数据流,在streaming中 以Dstream表示
    1)创建Dstream,对Dstream处理就行
    2)Dstream也是对rdd的再次封装,高级包装
    rdd的程序入口:创建sparkContext
    sparksql的程序入口:创建sparksession
    streaming的程序入口:创建sparkContext,构造streamingContext()

    范例

    在这里插入图片描述

    SparkStreaming与kafka的连接方式

    1)receive
    在这里插入图片描述
    2)直连方式
    在这里插入图片描述

    展开全文
  • 1. 什么是Spark Streaming Spark Streaming类似于Apache Strom, 用于流式数据的处理. 根据其官网解释, Spark Streaming有着高吞吐量和高容错的特点. Spark Streaming支持的数据源有很多, 例如:Kafka, Flume, ...

    1. 什么是Spark Streaming

    在这里插入图片描述

    • Spark Streaming类似于Apache Strom, 用于流式数据的处理.
    • 根据其官网解释, Spark Streaming有着高吞吐量高容错的特点.
    • Spark Streaming支持的数据源有很多, 例如:Kafka, Flume, Twitter, ZeroMQ和简单的TCP套接字等.
    • 数据输入后,可以使用Spark进行高度的抽象操作: map. flat, flatmap, window等进行运算,而且运算的结果能够保存到很多地方,比如hdfs, 数据库等.
    • Spark Streaming还能够和MLlib(机器学习)以及Graphx完美融合.
      在这里插入图片描述

    2. 为什么要学Spark Streaming

    2.1 易用

    在这里插入图片描述
    可以像写离线批处理一样写流式程序,支持java/scala/python语言.

    2.2 容错

    在这里插入图片描述
    SparkStreaming在没有额外代码和配置的情况下可以恢复丢失的工作。

    2.3 易整合到Spark体系

    在这里插入图片描述
    流式处理与批处理和交互式查询相结合。

    3. Spark和Strom有什么区别

    在这里插入图片描述

    • Storm擅长于动态处理大量实时生产的小数据块,概念上是将小数据量的数据源源不断传给过程;Spark擅长对现有的数据全集做处理,概念是将过程传给大数据量的数据。

    • 二者设计思路相反。Storm侧重于处理的实时性,Spark侧重处理庞大数据(类似于Hadoop的MR)。

    • Spark流模块(Spark Streaming)与Storm类似,但有区别:

      1.Storm纯实时,来一条数据,处理一条数据;SparkStreaming准实时,对一个时间段内的数据收集起来,作为一个RDD,再做处理。
      2.Storm响应时间毫秒级;Spark Streaming响应时间秒级
      3.Storm可以动态调整并行度;SparkStreaming不行

    • Storm应用场景:
      1.对于需要纯实时,不能忍受1秒以上延迟的场景
      2.要求可靠的事务机制和可靠性机制,即数据的处理完全精准
      3.如果还需要针对高峰低峰时间段,动态调整实时计算程序的并行度,以最大限度利用集群资源

    • Spark Streaming应用场景:
      1.Spark Streaming可以和Spark Core、Spark SQL无缝整合,如果一个项目除了实时计算之外,还包括了离线批处理、交互式查询等业务功能,考虑使用Spark Streaming。

    展开全文
  • 2、Spark Streaming特点易用容错易整合3、SparkStreaming与Storm的对比 1、Spark Streaming是什么? Spark Streaming类似于Apache Storm,用于流式数据的处理。 Spark Streaming有高吞吐量和容错能力强等特点。且...

    1、Spark Streaming是什么?

    Spark Streaming类似于Apache Storm,用于流式数据的处理。
    Spark Streaming有高吞吐量和容错能力强等特点。且支持的数据源有很多,例如:Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。
    数据输入后可以用Spark的高度抽象操作如:map、reduce、join、window等进行运算。而结果也能保存在很多地方,如HDFS,数据库等。另外Spark Streaming也能和MLlib(机器学习)以及Graphx完美融合。
    在这里插入图片描述

    官方地址:http://spark.apache.org/streaming/

    2、Spark Streaming特点

    易用

    可以像编写离线批处理一样去编写流式程序,支持java/scala/python语言。

    容错

    SparkStreaming在没有额外代码和配置的情况下可以恢复丢失的工作。
    它的容错性依赖于底层的rdd的容错机制。
    在这里插入图片描述

    易整合

    Spark Streaming 易整合到Spark体系。它可以将流式处理与批处理和交互式查询相结合。

    3、SparkStreaming与Storm的对比

    开发语言不同:spark Streaming的开发语言是Scala,Storm 的开发语言是Clojure.
    编程模型不同:Spark Streaming的编程模型是Dstream,Storm的编程模型是Spout/Bolt
    实时性不同:Spark Streaming是某一时间间隔的流式批量处理。实时性没有storm高,准实时处理。Storm是来一条数据处理一条,实时性比较高。

    展开全文
  • Spark StreamingSpark Streaming的使用

    千次阅读 2020-04-21 09:06:18
    一、Spark Streaming引入 集群监控 一般的大型集群和平台, 都需要对其进行监控的需求。 要针对各种数据库, 包括 MySQL, HBase 等进行监控 要针对应用进行监控, 例如 Tomcat, Nginx, Node.js 等 要针对硬件的...

    一、Spark Streaming引入

    集群监控

    一般的大型集群和平台, 都需要对其进行监控的需求。

    要针对各种数据库, 包括 MySQL, HBase 等进行监控

    要针对应用进行监控, 例如 Tomcat, Nginx, Node.js 等

    要针对硬件的一些指标进行监控, 例如 CPU, 内存, 磁盘 等

    Spark Streaming介绍

    官网:http://spark.apache.org/streaming/

    Spark Streaming是一个基于Spark Core之上的实时计算框架,可以从很多数据源消费数据并对数据进行实时的处理,具有高吞吐量和容错能力强等特点。

    Spark Streaming的特点

    1.易用

    可以像编写离线批处理一样去编写流式程序,支持java/scala/python语言。

    2.容错

    SparkStreaming在没有额外代码和配置的情况下可以恢复丢失的工作。

    3.易整合到Spark体系

    流式处理与批处理和交互式查询相结合。

    实时计算所处的位置

    二、Spark Streaming原理

    1、SparkStreaming原理

    整体流程

    Spark Streaming中,会有一个接收器组件Receiver,作为一个长期运行的task跑在一个Executor上。Receiver接收外部的数据流形成input DStream

    DStream会被按照时间间隔划分成一批一批的RDD,当批处理间隔缩短到秒级时,便可以用于处理实时数据流。时间间隔的大小可以由参数指定,一般设在500毫秒到几秒之间

    DStream进行操作就是对RDD进行操作,计算处理的结果可以传给外部系统。

    Spark Streaming的工作流程像下面的图所示一样,接到实时数据后,给数据分批次,然后传给Spark Engine(引擎)处理最后生成该批次的结果。

    数据抽象

    Spark Streaming的基础抽象是DStream(Discretized Stream,离散化数据流,连续不断的数据流),代表持续性的数据流经过各种Spark算子操作后的结果数据流

    可以从以下多个角度深入理解DStream

    1.DStream本质上就是一系列时间上连续的RDD

    2.DStream的数据的进行操作也是按照RDD为单位来进行的

    3.容错性

    底层RDD之间存在依赖关系,DStream直接也有依赖关系,RDD具有容错性,那么DStream也具有容错性

    如图:每一个椭圆形表示一个RDD

    椭圆形中的每个圆形代表一个RDD中的一个Partition分区

    每一列的多个RDD表示一个DStream(图中有三列所以有三个DStream)

    每一行最后一个RDD则表示每一个Batch Size所产生的中间结果RDD

    4.准实时性/近实时性

    Spark Streaming将流式计算分解成多个Spark Job,对于每一时间段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。

    对于目前版本的Spark Streaming而言,其最小的Batch Size的选取在0.5~5秒钟之间

    所以Spark Streaming能够满足流式准实时计算场景,对实时性要求非常高的如高频实时交易场景则不太适合

    总结       

    简单来说DStream就是对RDD的封装,你对DStream进行操作,就是对RDD进行操作

    对于DataFrame/DataSet/DStream来说本质上都可以理解成RDD

    2、DStream相关操作

    DStream上的操作与RDD的类似,分为以下两种:

    Transformations(转换)

    Output Operations(输出)/Action

    Transformations

    常见Transformation---无状态转换:每个批次的处理不依赖于之前批次的数据

    Transformation

    Meaning

    map(func)

    对DStream中的各个元素进行func函数操作,然后返回一个新的DStream

    flatMap(func)

    与map方法类似,只不过各个输入项可以被输出为零个或多个输出项

    filter(func)

    过滤出所有函数func返回值为true的DStream元素并返回一个新的DStream

    union(otherStream)

    将源DStream和输入参数为otherDStream的元素合并,并返回一个新的DStream.

    reduceByKey(func, [numTasks])

    利用func函数对源DStream中的key进行聚合操作,然后返回新的(K,V)对构成的DStream

    join(otherStream, [numTasks])

    输入为(K,V)(K,W)类型的DStream,返回一个新的(K(V,W)类型的DStream

    transform(func)

    通过RDD-to-RDD函数作用于DStream中的各个RDD,可以是任意的RDD操作,从而返回一个新的RDD

    特殊的Transformations---有状态转换:当前批次的处理需要使用之前批次的数据或者中间结果

    有状态转换包括基于追踪状态变化的转换(updateStateByKey)和滑动窗口的转换

    1.UpdateStateByKey(func)

    2.Window Operations 窗口操作

    Output/Action

    Output Operations可以DStream的数据输出到外部的数据库或文件系统

    当某个Output Operations被调用时,spark streaming程序才会开始真正的计算过程(与RDD的Action类似)

    Output Operation

    Meaning

    print()

    打印到控制台

    saveAsTextFiles(prefix, [suffix])

    保存流的内容为文本文件,文件名为"prefix-TIME_IN_MS[.suffix]".

    saveAsObjectFiles(prefix,[suffix])

    保存流的内容为SequenceFile,文件名为 "prefix-TIME_IN_MS[.suffix]".

    saveAsHadoopFiles(prefix,[suffix])

    保存流的内容为hadoop文件,文件名为"prefix-TIME_IN_MS[.suffix]".

    foreachRDD(func)

    对Dstream里面的每个RDD执行func

    总结

    三、Spark Streaming实战

    1、WordCount

    需求&准备

    首先在linux服务器上安装nc工具

    nc是netcat的简称,原本是用来设置路由器,我们可以利用它向某个端口发送数据

    yum install -y nc

    ●启动一个服务端并开放9999端口,等一下往这个端口发数据

    nc -lk 9999

    ●发送数据

    import org.apache.spark.sql.catalyst.expressions.Second
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Demo01_WC {
    
      def main(args: Array[String]): Unit = {
    
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WC")
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
        //监听socket接收数据
        val dataDStrem: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)
        //操作数据
        dataDStrem.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

    执行

    1.先执行nc -lk 9999

    2.然后执行代码

    3.不断的在1中输入不同的单词

    hadoop spark sqoop hadoop spark hive hadoop

    4.观察IDEA控制台输出

    sparkStreaming每隔5s计算一次当前5s内的数据,然后将每个批次的数据输出

    2、updateStateByKey

    在上面的那个案例中存在这样一个问题:

    每个批次的单词次数都被正确的统计出来,但是结果不能累加!

    如果需要累加需要使用updateStateByKey(func)来更新状态

    import org.apache.spark.streaming.dstream.ReceiverInputDStream
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Demo02_updateStateByKey {
    
      def main(args: Array[String]): Unit = {
    
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WC")
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
        //历史数据存在的目录
        ssc.checkpoint("./wc")
        //监听socket接收数据
        val dataDStrem: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)
        //操作数据
        dataDStrem.flatMap(_.split(" ")).map((_,1)).updateStateByKey(updateFunc).print()
        ssc.start()
        ssc.awaitTermination()
      }
      def updateFunc(currentValues:Seq[Int],historyValue:Option[Int])={
        //currentValues当前值
        //historyValue历史值
        val result: Int = currentValues.sum+historyValue.getOrElse(0)
        Some(result)
      }
    
    }

     执行

    1.先执行nc -lk 9999

    2.然后执行以上代码

    3.不断的在1中输入不同的单词,

    hadoop spark sqoop hadoop spark hive hadoop

    4.观察IDEA控制台输出

    sparkStreaming每隔5s计算一次当前5s内的数据,然后将每个批次的结果数据累加输出。

    3、reduceByKeyAndWindow

    图解

    滑动窗口转换操作的计算过程如下图所示,

    我们可以事先设定一个滑动窗口的长度(也就是窗口的持续时间),并且设定滑动窗口的时间间隔(每隔多长时间执行一次计算),

    比如设置滑动窗口的长度(也就是窗口的持续时间)为24H,设置滑动窗口的时间间隔(每隔多长时间执行一次计算)为1H

    那么意思就是:每隔1H计算最近24H的数据

    代码演示

    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Demo03_reduceByKeyAndWindow {
    
      def main(args: Array[String]): Unit = {
    
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WC")
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
        //历史数据存在的目录
        ssc.checkpoint("./wc")
        //监听socket接收数据
        val dataDStrem: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)
        //操作数据
        val wordAndOneDStream: DStream[(String, Int)] = dataDStrem.flatMap(_.split(" ")).map((_,1))
        //使用窗口函数进行WordCount计数
        wordAndOneDStream.reduceByKeyAndWindow((a:Int,b:Int)=>a+b,Seconds(10),Seconds(5)).print()
        ssc.start()
        ssc.awaitTermination()
      }
      def updateFunc(currentValues:Seq[Int],historyValue:Option[Int])={
        //currentValues当前值
        //historyValue历史值
        val result: Int = currentValues.sum+historyValue.getOrElse(0)
        Some(result)
      }
    
    }

    执行

    1.先执行nc -lk 9999

    2.然后执行以上代码

    3.不断的在1中输入不同的单词

    hadoop spark sqoop hadoop spark hive hadoop

    4.观察IDEA控制台输出

    现象:sparkStreaming每隔5s计算一次当前在窗口大小为10s内的数据,然后将结果数据输出。

    4、统计一定时间内的热门词汇TopN

    需求

    模拟百度热搜排行榜

    统计最近10s的热搜词Top3,每隔5秒计算一次

    WindowDuration = 10s

    SlideDuration = 5s

    代码演示

    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
     /*
      * Desc 我们要模拟百度热搜排行榜统计最近10s的热搜词Top3,每隔5秒计算一次
      */
    object WordCount4 {
      def main(args: Array[String]): Unit = {
        //1.创建StreamingContext
        //spark.master should be set as local[n], n > 1
        val conf = new SparkConf().setAppName("wc").setMaster("local[*]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD
        //2.监听Socket接收数据
        //ReceiverInputDStream就是接收到的所有的数据组成的RDD,封装成了DStream,接下来对DStream进行操作就是对RDD进行操作
        val dataDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)
        //3.操作数据
        val wordDStream: DStream[String] = dataDStream.flatMap(_.split(" "))
        val wordAndOneDStream: DStream[(String, Int)] = wordDStream.map((_,1))
        //4.使用窗口函数进行WordCount计数
        val wordAndCount: DStream[(String, Int)] = wordAndOneDStream.reduceByKeyAndWindow((a:Int,b:Int)=>a+b,Seconds(10),Seconds(5))
        val sorteDStream: DStream[(String, Int)] = wordAndCount.transform(rdd => {
          val sortedRDD: RDD[(String, Int)] = rdd.sortBy(_._2, false) //逆序/降序
          println("===============top3==============")
          sortedRDD.take(3).foreach(println)
          println("===============top3==============")
          sortedRDD
        }
        )
        //No output operations registered, so nothing to execute
        sorteDStream.print
        ssc.start()//开启
        ssc.awaitTermination()//等待优雅停止
      }
    }

    执行

    1.先执行nc -lk 9999

    2.然后在执行以上代码

    3.不断的在1中输入不同的单词

    hadoop spark sqoop hadoop spark hive hadoop

    4.观察IDEA控制台输出

    四、整合kafka

    1、Kafka快速回顾

    Broker : 安装Kafka服务的机器就是一个broker

    Producer :消息的生产者,负责将数据写入到broker中(push)

    Consumer:消息的消费者,负责从kafka中拉取数据(pull),老版本的消费者需要依赖zk,新版本的不需要

    Topic: 主题,相当于是数据的一个分类,不同topic存放不同业务的数据 --主题:区分业务

    Replication:副本,数据保存多少份(保证数据不丢失) --副本:数据安全

    Partition:分区,是一个物理的分区,一个分区就是一个文件,一个Topic可以有1~n个分区,每个分区都有自己的副本 --分区:并发读写

    Consumer Group:消费者组,一个topic可以有多个消费者/组同时消费,多个消费者如果在一个消费者组中,那么他们不能重复消费数据 --消费者组:提高消费者消费速度、方便统一管理

    注意:一个Topic可以被多个消费者或者组订阅,一个消费者/组也可以订阅多个主题

    注意:读数据只能从Leader读, 写数据也只能往Leader写,Follower会从Leader那里同步数据过来做副本!!!

    常用命令
    #启动kafka
    /export/servers/kafka/bin/kafka-server-start.sh -daemon /export/servers/kafka/config/server.properties 

    #停止kafka
    /export/servers/kafka/bin/kafka-server-stop.sh 

    #查看topic信息
    /export/servers/kafka/bin/kafka-topics.sh --list --zookeeper node01:2181
     
    #创建topic
    /export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 3 --partitions 3 --topic test
     
    #查看某个topic信息
    /export/servers/kafka/bin/kafka-topics.sh --describe --zookeeper node01:2181 --topic test
     
    #删除topic
    /export/servers/kafka/bin/kafka-topics.sh --zookeeper node01:2181 --delete --topic test
     
    #启动生产者--控制台的生产者一般用于测试
    /export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092 --topic spark_kafka
     
    #启动消费者--控制台的消费者一般用于测试
    /export/servers/kafka/bin/kafka-console-consumer.sh --zookeeper node01:2181 --topic spark_kafka--from-beginning
     
    # 消费者连接到borker的地址
    /export/servers/kafka/bin/kafka-console-consumer.sh --bootstrap-server node01:9092,node02:9092,node03:9092 --topic spark_kafka --from-beginning

    2、整合Kafka两种模式说明

    面试题:Receiver & Direct

    开发中我们经常会利用SparkStreaming实时地读取kafka中的数据然后进行处理,在spark1.3版本后,kafkaUtils里面提供了两种创建DStream的方法:

    1.Receiver接收方式:

    KafkaUtils.createDstream(开发中不用,了解即可,但是面试可能会问)

    Receiver作为常驻的Task运行在Executor等待数据,但是一个Receiver效率低,需要开启多个,再手动合并数据(union),再进行处理,很麻烦

    Receiver哪台机器挂了,可能会丢失数据,所以需要开启WAL(预写日志)保证数据安全,那么效率又会降低!

    Receiver方式是通过zookeeper来连接kafka队列,调用Kafka高阶API,offset存储在zookeeper,由Receiver维护,

    spark在消费的时候为了保证数据不丢也会在Checkpoint中存一份offset,可能会出现数据不一致

    所以不管从何种角度来说,Receiver模式都不适合在开发中使用了,已经淘汰了

    2.Direct直连方式:

    KafkaUtils.createDirectStream(开发中使用,要求掌握)

    Direct方式是直接连接kafka分区来获取数据,从每个分区直接读取数据大大提高了并行能力

    Direct方式调用Kafka低阶API(底层API),offset自己存储和维护,默认由Spark维护在checkpoint中,消除了与zk不一致的情况 

    当然也可以自己手动维护,把offset存在mysql、redis中

    所以基于Direct模式可以在开发中使用,且借助Direct模式的特点+手动操作可以保证数据的Exactly once 精准一次

     总结:

    Receiver接收方式

    1. 多个Receiver接受数据效率高,但有丢失数据的风险。
    2. 开启日志(WAL)可防止数据丢失,但写两遍数据效率低。
    3. Zookeeper维护offset有重复消费数据可能。
    4. 使用高层次的API

    Direct直连方式

    1. 不使用Receiver,直接到kafka分区中读取数据
    2. 不使用日志(WAL)机制。
    3. Spark自己维护offset
    4. 使用低层次的API

    扩展:关于消息语义

    实现方式

    消息语义

    存在的问题

    Receiver

    at most once

    最多被处理一次

    会丢失数据

    Receiver+WAL

    at least once

    最少被处理一次

    不会丢失数据,但可能会重复消费,且效率低

    Direct+手动操作

    exactly once

    只被处理一次/精准一次

    不会丢失数据,也不会重复消费,且效率高

    ●注意:

    开发中SparkStreaming和kafka集成有两个版本:0.8及0.10+

    0.8版本有Receiver和Direct模式(但是0.8版本生产环境问题较多,在Spark2.3之后不支持0.8版本了)

    0.10以后只保留了direct模式(Reveiver模式不适合生产环境),并且0.10版本API有变化(更加强大)

    3、spark-streaming-kafka-0-8(了解)

    Receiver

    KafkaUtils.createDstream使用了receivers来接收数据,利用的是Kafka高层次的消费者api,偏移量由Receiver维护在zk中,对于所有的receivers接收到的数据将会保存在Spark executors中,然后通过Spark Streaming启动job来处理这些数据,默认会丢失,可启用WAL日志,它同步将接受到数据保存到分布式文件系统上比如HDFS。保证数据在出错的情况下可以恢复出来。尽管这种方式配合着WAL机制可以保证数据零丢失的高可靠性,但是启用了WAL效率会较低,且无法保证数据被处理一次且仅一次,可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。

    官方现在已经不推荐这种整合方式

    准备工作

    1.启动zookeeper集群

    zkServer.sh start

    2.启动kafka集群

    kafka-server-start.sh  /export/servers/kafka/config/server.properties

    3.创建topic

    kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 3 --topic spark_kafka

    4.通过shell命令向topic发送消息

    kafka-console-producer.sh --broker-list node01:9092 --topic  spark_kafka

    hadoop spark sqoop hadoop spark hive hadoop

    5.添加kafka的pom依赖

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId> spark-streaming-kafka-0-10-assembly_2.11
     </artifactId>
        <version>2.2.0</version>
    </dependency>

    API

    通过receiver接收器获取kafka中topic数据,可以并行运行更多的接收器读取kafak topic中的数据,这里为3个

        val receiverDStream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {

          val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)

          stream

        })

    如果启用了WAL(spark.streaming.receiver.writeAheadLog.enable=true)可以设置存储级别(默认StorageLevel.MEMORY_AND_DISK_SER_2)

    代码演示

    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    import scala.collection.immutable
    
    object SparkKafka {
      def main(args: Array[String]): Unit = {
        //1.创建StreamingContext
        val config: SparkConf = 
    new SparkConf().setAppName("SparkStream").setMaster("local[*]")
          .set("spark.streaming.receiver.writeAheadLog.enable", "true")
    //开启WAL预写日志,保证数据源端可靠性
        val sc = new SparkContext(config)
        sc.setLogLevel("WARN")
        val ssc = new StreamingContext(sc,Seconds(5))
        ssc.checkpoint("./kafka")
    //==============================================
        //2.准备配置参数
        val zkQuorum = "node01:2181,node02:2181,node03:2181"
        val groupId = "spark"
        val topics = Map("spark_kafka" -> 2)//2表示每一个topic对应分区都采用2个线程去消费,
    //ssc的rdd分区和kafka的topic分区不一样,增加消费线程数,并不增加spark的并行处理数据数量
        //3.通过receiver接收器获取kafka中topic数据,可以并行运行更多的接收器读取kafak topic中的数据,这里为3个
        val receiverDStream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {
          val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)
          stream
        })
        //4.使用union方法,将所有receiver接受器产生的Dstream进行合并
        val allDStream: DStream[(String, String)] = ssc.union(receiverDStream)
        //5.获取topic的数据(String, String) 第1个String表示topic的名称,第2个String表示topic的数据
        val data: DStream[String] = allDStream.map(_._2)
    //==============================================
        //6.WordCount
        val words: DStream[String] = data.flatMap(_.split(" "))
        val wordAndOne: DStream[(String, Int)] = words.map((_, 1))
        val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _)
        result.print()
        ssc.start()
        ssc.awaitTermination()
      }
    }

    Direct

    Direct方式会定期地从kafka的topic下对应的partition中查询最新的偏移量,再根据偏移量范围在每个batch里面处理数据,Spark通过调用kafka简单的消费者API读取一定范围的数据。

    • Direct的缺点是无法使用基于zookeeper的kafka监控工具
    • Direct相比基于Receiver方式有几个优点: 
    • 简化并行

    不需要创建多个kafka输入流,然后union它们,sparkStreaming将会创建和kafka分区数一样的rdd的分区数,而且会从kafka中并行读取数据,sparkRDD的分区数和kafka中的分区数据是一一对应的关系。

    • 高效

    Receiver实现数据的零丢失是将数据预先保存在WAL中,会复制一遍数据,会导致数据被拷贝两次,第一次是被kafka复制,另一次是写到WAL中。而Direct不使用WAL消除了这个问题。 

    • 恰好一次语义(Exactly-once-semantics)

    Receiver读取kafka数据是通过kafka高层次api把偏移量写入zookeeper中,虽然这种方法可以通过数据保存在WAL中保证数据不丢失,但是可能会因为sparkStreaming和ZK中保存的偏移量不一致而导致数据被消费了多次。

    Direct的Exactly-once-semantics(EOS)通过实现kafka低层次api,偏移量仅仅被ssc保存在checkpoint中,消除了zk和ssc偏移量不一致的问题。

    API

    KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

    代码演示

    import kafka.serializer.StringDecoder
    import org.apache.spark.streaming.dstream.{DStream, InputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    
    
    object SparkKafka2 {
      def main(args: Array[String]): Unit = {
        //1.创建StreamingContext
        val config: SparkConf = 
    new SparkConf().setAppName("SparkStream").setMaster("local[*]")
        val sc = new SparkContext(config)
        sc.setLogLevel("WARN")
        val ssc = new StreamingContext(sc,Seconds(5))
        ssc.checkpoint("./kafka")
        //==============================================
        //2.准备配置参数
        val kafkaParams = Map("metadata.broker.list" -> "node01:9092,node02:9092,node03:9092", "group.id" -> "spark")
        val topics = Set("spark_kafka")
        val allDStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
        //3.获取topic的数据
        val data: DStream[String] = allDStream.map(_._2)
        //==============================================
        //WordCount
        val words: DStream[String] = data.flatMap(_.split(" "))
        val wordAndOne: DStream[(String, Int)] = words.map((_, 1))
        val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _)
        result.print()
        ssc.start()
        ssc.awaitTermination()
      }
    }

    4/spark-streaming-kafka-0-10

    spark-streaming-kafka-0-10版本中,API有一定的变化,操作更加灵活,开发中使用

    pom.xml

    <!--<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>-->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>

    API

    http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

    创建topic

    /export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 3 --partitions 3 --topic spark_kafka

    启动生产者

    /export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092,node01:9092,node01:9092 --topic spark_kafka

    代码演示

    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.streaming.dstream.{DStream, InputDStream}
    import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    object Demo05_SparkKafka {
    
      def main(args: Array[String]): Unit = {
        //创建sparkConf
        val conf: SparkConf = new SparkConf().setAppName("wc").setMaster("local[*]")
        //创建sparkContext
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("WARN")
        //
        val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))
    
        val kafkaParams = Map[String, Object](
          "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> "SparkKafkaDemo",
          //earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
          //latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
          //none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
          //这里配置latest自动重置偏移量为最新的偏移量,即如果有偏移量从偏移量位置开始消费,没有偏移量从新来的数据开始消费
          "auto.offset.reset" -> "latest",
          //false表示关闭自动提交.由spark帮你提交到Checkpoint或程序员手动维护
          "enable.auto.commit" -> (false: java.lang.Boolean)
        )
    //    val topics = Array("spark_kafka")
    //    val recordDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](Array("spark_kafka"), kafkaParams))
           val kafkaDatas: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](Array("spark_kafka"), kafkaParams))
        val wordOne: DStream[(String, Int)] = kafkaDatas.flatMap(x=>x.value().split(" ")).map((_,1))
        val wordAndOneDStream: DStream[(String, Int)] = wordOne.reduceByKeyAndWindow((a:Int, b:Int)=>a+b,Seconds(10),Seconds(5))
    
        val result: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(_+_)
        result.print()
        ssc.start()//开启
        ssc.awaitTermination()//等待优雅停止
      }
    }

    5、扩展:Kafka手动维护偏移量

    API

    http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

    启动生产者

    /export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092,node01:9092,node01:9092 --topic spark_kafka

    代码演示

    import java.sql.{DriverManager, ResultSet}
    
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.kafka.common.TopicPartition
    import org.apache.kafka.common.serialization.StringDeserializer
    import org.apache.spark.streaming.dstream.InputDStream
    import org.apache.spark.streaming.kafka010.{OffsetRange, _}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.{SparkConf, SparkContext}
    import scala.collection.mutable
    
    object SparkKafkaDemo2 {
      def main(args: Array[String]): Unit = {
        //1.创建StreamingContext
        //spark.master should be set as local[n], n > 1
        val conf = new SparkConf().setAppName("wc").setMaster("local[*]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD
        //准备连接Kafka的参数
        val kafkaParams = Map[String, Object](
          "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> "SparkKafkaDemo",
          "auto.offset.reset" -> "latest",
          "enable.auto.commit" -> (false: java.lang.Boolean)
        )
        val topics = Array("spark_kafka")
        //2.使用KafkaUtil连接Kafak获取数据
        //注意:
        //如果MySQL中没有记录offset,则直接连接,从latest开始消费
        //如果MySQL中有记录offset,则应该从该offset处开始消费
        val offsetMap: mutable.Map[TopicPartition, Long] = OffsetUtil.getOffsetMap("SparkKafkaDemo","spark_kafka")
        val recordDStream: InputDStream[ConsumerRecord[String, String]] = if(offsetMap.size > 0){//有记录offset
          println("MySQL中记录了offset,则从该offset处开始消费")
          KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,//位置策略,源码强烈推荐使用该策略,会让Spark的Executor和Kafka的Broker均匀对应
          ConsumerStrategies.Subscribe[String, String](topics, kafkaParams,offsetMap))//消费策略,源码强烈推荐使用该策略
        }else{//没有记录offset
          println("没有记录offset,则直接连接,从latest开始消费")
          // /export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092 --topic  spark_kafka
          KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,//位置策略,源码强烈推荐使用该策略,会让Spark的Executor和Kafka的Broker均匀对应
          ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))//消费策略,源码强烈推荐使用该策略
        }
        //3.操作数据
        //注意:我们的目标是要自己手动维护偏移量,也就意味着,消费了一小批数据就应该提交一次offset
        //而这一小批数据在DStream的表现形式就是RDD,所以我们需要对DStream中的RDD进行操作
        //而对DStream中的RDD进行操作的API有transform(转换)和foreachRDD(动作)
        recordDStream.foreachRDD(rdd=>{
          if(rdd.count() > 0){//当前这一时间批次有数据
            rdd.foreach(record => println("接收到的Kafk发送过来的数据为:" + record))
            //接收到的Kafk发送过来的数据为:ConsumerRecord(topic = spark_kafka, partition = 1, offset = 6, CreateTime = 1565400670211, checksum = 1551891492, serialized key size = -1, serialized value size = 43, key = null, value = hadoop spark ...)
            //注意:通过打印接收到的消息可以看到,里面有我们需要维护的offset,和要处理的数据
            //接下来可以对数据进行处理....或者使用transform返回和之前一样处理
            //处理数据的代码写完了,就该维护offset了,那么为了方便我们对offset的维护/管理,spark提供了一个类,帮我们封装offset的数据
            val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
            for (o <- offsetRanges){
              println(s"topic=${o.topic},partition=${o.partition},fromOffset=${o.fromOffset},untilOffset=${o.untilOffset}")
            }
            //手动提交offset,默认提交到Checkpoint中
            //recordDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
            //实际中偏移量可以提交到MySQL/Redis中
            OffsetUtil.saveOffsetRanges("SparkKafkaDemo",offsetRanges)
          }
        })
    
       /* val lineDStream: DStream[String] = recordDStream.map(_.value())//_指的是ConsumerRecord
        val wrodDStream: DStream[String] = lineDStream.flatMap(_.split(" ")) //_指的是发过来的value,即一行数据
        val wordAndOneDStream: DStream[(String, Int)] = wrodDStream.map((_,1))
        val result: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(_+_)
        result.print()*/
        ssc.start()//开启
        ssc.awaitTermination()//等待优雅停止
      }
    
      /*
      手动维护offset的工具类
      首先在MySQL创建如下表
        CREATE TABLE `t_offset` (
          `topic` varchar(255) NOT NULL,
          `partition` int(11) NOT NULL,
          `groupid` varchar(255) NOT NULL,
          `offset` bigint(20) DEFAULT NULL,
          PRIMARY KEY (`topic`,`partition`,`groupid`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
       */
      object OffsetUtil {
    
        //从数据库读取偏移量
        def getOffsetMap(groupid: String, topic: String) = {
          val connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")
          val pstmt = connection.prepareStatement("select * from t_offset where groupid=? and topic=?")
          pstmt.setString(1, groupid)
          pstmt.setString(2, topic)
          val rs: ResultSet = pstmt.executeQuery()
          val offsetMap = mutable.Map[TopicPartition, Long]()
          while (rs.next()) {
            offsetMap += new TopicPartition(rs.getString("topic"), rs.getInt("partition")) -> rs.getLong("offset")
          }
          rs.close()
          pstmt.close()
          connection.close()
          offsetMap
        }
    
        //将偏移量保存到数据库
        def saveOffsetRanges(groupid: String, offsetRange: Array[OffsetRange]) = {
          val connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")
          //replace into表示之前有就替换,没有就插入
          val pstmt = connection.prepareStatement("replace into t_offset (`topic`, `partition`, `groupid`, `offset`) values(?,?,?,?)")
          for (o <- offsetRange) {
            pstmt.setString(1, o.topic)
            pstmt.setInt(2, o.partition)
            pstmt.setString(3, groupid)
            pstmt.setLong(4, o.untilOffset)
            pstmt.executeUpdate()
          }
          pstmt.close()
          connection.close()
        }
      }
    }

     

    展开全文
  • 支持动态调节 streaming 的 批次间隔时间 (不同于sparkstreaming 的 定长的批次间隔,StructuredStreaming中使用trigger实现了。) 支持在streaming过程中 重设 topics,用于生产中动态地增加删减数据源 添加了速率...
  • SparkStreaming接收Kafka数据的两种方式一、SparkStreaming + Kafka Receiver模式二、SparkStreaming + Kafka Direct模式三、Direct模式与Receiver模式比较四、SparkStreaming+Kafka维护消费者offset 一、Spark...
  • 基于 Flume+ Kafka+ Spark Streaming 实现实时监控输出日志的报警系统的 Spark Streaming 程序代码。 基于 Flume+Kafka+Spark Streaming 实现实时监控输出日志的报警系统的 Spark Streaming 程序代码,博客链接: ...
  • Spark Streaming原理 Spark Streaming 是基于spark的流式批处理引擎,其基本原理是把输入数据以某一时间间隔批量的处理,当批处理间隔缩短到秒级时,便可以用于处理实时数据流。 Spark Streaming计算流程 Spark ...
  • SparkStreaming简单介绍 流式数据处理(stream processing) 要处理的数据就像流水一样,源源不断的产生数据,需要实时进行处理 对SparkCore的高级API的封装,将流式的数据切分为小的批次batch(按照时间间隔)的...
  • java-sparkstreaming-kinesis 这个项目是用Java运行一个Spark Streaming应用程序 mvn clean compile assembly:assembly >> 使用 deps 构建 jar spark-submit --class ...
  • 7.SparkStreaming(上)--SparkStreaming原理介绍.pdf 7.SparkStreaming(下)--SparkStreaming实战.pdf 8.SparkMLlib(上)--机器学习及SparkMLlib简介.pdf 8.SparkMLlib(下)--SparkMLlib实战.pdf 9.SparkGraphX...
  • SparkStreaming编程讲解

    2021-01-30 03:27:35
    1.什么是SparkStreaming?2.SparkStreaming可以接受那些数据源?3.Dstream,我们可以进行哪两种操作?Sparkstreaming:构建在Spark上处理Stream数据的框架,基本的原理是将Stream数据分成小的时间片断(几秒),以...
  • 从今天起,我们踏上了新的Spark学习旅途。我们的目标是要像Spark官方机构那样有能力去定制Spark。...Spark的子框架已有若干,为何选择Spark Streaming?让我们细细道来。  Spark最开始只有Spark Cor
  • spark Streaming和structed streaming分析,理解整个 Spark Streaming 的模块划分和代码逻辑。
  • 基于 Flume+Kafka+Spark Streaming 实现实时监控输出日志的报警系统的 Spark Streaming 程序代码,博客链接: https://blog.csdn.net/linge1995/article/details/81326146
  • Spark Streaming简介

    2021-01-07 03:48:50
    Spark StreamingSpark 核心 API 的一个扩展,可以实现高吞吐量的、具备容错机制的实时流数据的处理。 Spark Streaming 支持从多种数据源获取数据,包括 Kafka、Flume、Twitter、ZeroMQ、Kinesis 以及 TCP ...
  • sparkstreaming.zip

    2019-11-27 10:24:48
    java的sparkstreaming连接kafka的例子,kafka生产者生产消息,消费者读取消息,sparkstreaming读取kafka小区并进行存储iotdb数据库。

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 50,933
精华内容 20,373
关键字:

sparkstreaming