2018-06-07 00:51:11 lyzx_in_csdn 阅读数 2836
  • Spark开发工程师(含项目)

    本课程为大数据金融信贷项目实战课,着重讲解企业中常用的大数据技术理论与实战,如Hadoop、Hive、HBase、Sqoop、Flume、Kafka、Spark Streaming、Spark SQL、Spark Structured Streaming等。课程包含离线项目和实时项目,从项目业务需求、技术选型、架构设计、集群安装部署、集成开发以及项目可视化进行全方位实战讲解。

    1690 人正在学习 去看看 李飞

spark和kafka整合有2中方式

1、receiver

顾名思义:就是有一个线程负责获取数据,这个线程叫receiver线程

解释:

1、Spark集群中的某个executor中有一个receiver线程,这个线程负责从kafka中获取数据

 注意:这里的获取数据并不是从kafka中拉(pull) 而是接收数据,具体原理是该receiver线程发送请求到kafka,这个请求包含对kafka中每个partition的消费偏移量(offset),然后由kafka主动的推送数据到spark中,再有该receiver线程负责接收数据

2、当receiver线程接收到数据后会做备份处理,即把数据备份到其他的executor中,也可能会备份到这个receiver线程所在节点的executor中

3、当备份完毕后该线程会把每个partition的消费偏移量在zookeeper中修改,(新版本的kafka的offset 保存在kafka集群中)

4、修改完offset后,该receiver线程会把"消费"的数据告诉Driver

5、Driver分发任务时会根据每个executor上的数据,根据数据本地性发送

问题:

当第三步执行完后,对于kafka来说这一批数据已经消费完成,那么如果此时Driver挂掉,那么这一批数据就会丢失,为了解决这个问题,有一个叫WAL逾写日志的概念,即把一部分数据存储在HDFS上,当Driver回复后可以从HDFS上获取这部分数据,但是开启WAL性能会受到很大的影响

 

2、dirct

直接连接:即每个executor直接取kafka获取数据

1、首先Driver程序会定时(batchInterval)的向executor中发送任务(4个)

  >> 问题1:Driver怎么知道要把任务发送到哪个executor中呢?

     >> Driver会调用Kafka的接口获取某个partition位于哪个节点上,根据这个来获取这些信息并发送任务到指定的节点,这就类似于Spark集群处理HDFS上的文件数据,Spark是可以知道某些文件的block在那些节点上,就是spark调用了HDFS的相关接口

  >> 问题2:为什么是4个任务?

      >> 这个个数由消费的topic的partition的个数决定,因为spark会对每个partition开启一个任务,所以任务数是kafka的某个topic的partition数

2、当每个任务确定了处理那个partition中的数据,则就有任务本身去kafka获取数据

总结:目前公司中第二种方式使用比较多,这样也有一个问题就是说当kafka中某个topic加了ACL验证,那么这种方式是不能消费加了ACL的topic中的数据,因为kafka客户端的ACL验证需要客户端配置一个环境变量在System的Properties中,在local模式下可以实现,因为local模式下启动一个虚拟机实例,即只对应一个System,而在集群模式下,要启动多个进程,即启动多个虚拟机实例,所以System的全局属性没有办法配置

2019-03-14 18:05:21 zimiao552147572 阅读数 533
  • Spark开发工程师(含项目)

    本课程为大数据金融信贷项目实战课,着重讲解企业中常用的大数据技术理论与实战,如Hadoop、Hive、HBase、Sqoop、Flume、Kafka、Spark Streaming、Spark SQL、Spark Structured Streaming等。课程包含离线项目和实时项目,从项目业务需求、技术选型、架构设计、集群安装部署、集成开发以及项目可视化进行全方位实战讲解。

    1690 人正在学习 去看看 李飞

 


KafkaUtils.createDirectStream方式(推荐使用createDirectStream,不推荐使用createDstream)
        1.createDirectStream方式不同于createDstream的Receiver接收数据的方式,createDirectStream定期地从kafka的topic下对应的partition中查询最新的偏移量,
          再根据偏移量范围在每个batch(批次)里面处理数据,Spark通过调用kafka简单的消费者Api(低级api)读取一定范围的数据。
        2.createDirectStream方式相比基于Receiver的createDstream方式有几个优点: 
            1.简化并行
                不需要创建多个kafka输入流,然后union它们,sparkStreaming将会创建和kafka中的topic分区数相同的rdd的分区数,
                而且会从kafka中并行读取数据,spark中RDD的分区数和kafka中的topic分区数是一一对应的关系。
            2.高效     
                createDstream方式实现数据的零丢失是将数据预先保存在WAL中,会复制一遍数据,会导致数据被拷贝两次,
                第一次是接受kafka中topic的数据,另一次是写到WAL中。而不使用createDstream的receiver的这种方式则消除了这个问题。 
            3.恰好一次语义(Exactly-once-semantics,即EOS)
                createDstream方式的Receiver读取kafka数据是通过 kafka的高级api 把偏移量写入zookeeper中,
                虽然这种方法可以通过把数据保存在WAL中保证数据不丢失,但是可能会因为Spark Streaming和zookeeper中保存的偏移量不一致
                而导致数据被消费了多次。
                createDirectStream方式的恰好一次语义(Exactly-once-semantics,即EOS) 则通过实现 kafka低级api,
                topic中的偏移量仅仅被StreamingContext保存在客户端程序的checkpoint目录中,topic中的偏移量则不再保存在zookeeper中,
                那么程序只需要到checkpoint目录中读取topic中的偏移量即可,便从而消除了zookeeper和StreamingContext中偏移量不一致的问题,
                但是同时也造成无法使用基于zookeeper的kafka监控工具来读取到topic的相关信息。

        3.编写Spark Streaming应用程序(createDirectStream方式的版本)
            package cn.itcast.dstream.kafka
            import kafka.serializer.StringDecoder
            import org.apache.spark.{SparkConf, SparkContext}
            import org.apache.spark.streaming.{Seconds, StreamingContext}
            import org.apache.spark.streaming.dstream.{DStream, InputDStream}
            import org.apache.spark.streaming.kafka.KafkaUtils

            // 利用sparkStreaming对接kafka实现单词计数:createDirectStream 采用Direct(低级API)
            object SparkStreamingKafka_Direct 
            {
                    def main(args: Array[String]): Unit = 
                {
                          //1、创建sparkConf。setMaster("local[2]"):本地模式运行 
                    //setMaster("local[2]"):本地模式运行,启动两个线程
                    //设置master的地址local[N] ,n必须大于1,其中1个线程负责去接受数据,另一线程负责处理接受到的数据
                            val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingKafka_Direct").setMaster("local[2]")
                            //2、创建sparkContext
                            val sc = new SparkContext(sparkConf)
                    //设置日志输出的级别
                            sc.setLogLevel("WARN")
                        //3、构建StreamingContext对象,每个批处理的时间间隔,即以多少秒内的数据划分为一个批次 ,当前设置 以5秒内的数据 划分为一个批次  
                            val ssc = new StreamingContext(sc, Seconds(5))
                    //createDirectStream采用低级API的关系,topic中的偏移量仅仅被StreamingContext保存在客户端程序的checkpoint目录中,
                       //topic中的偏移量则不再保存在zookeeper中。此处设置checkpoint路径,当前项目下有一个Kafka_Direct目录
                          ssc.checkpoint("./Kafka_Direct")
                            //4、配置kafka相关参数:配置每个kafka服务器的IP:端口,还有"group.id"(消费者组groupId)
                    //配置group.id:多个Consumer的group.id都相同的话,表示多个Consumer都在同一个消费组group中
                            val kafkaParams=Map("metadata.broker.list"->"NODE1:9092,NODE2:9092,NODE3:9092","group.id"->"Kafka_Direct")
                            //5、定义topic的主题名。
                    //此处没使用topics.split("分隔符").toSet。topic中数据还可设置以分隔符进行切割,进而切割出每行数据
                            val topics=Set("kafka_spark")
                            //6、通过 KafkaUtils.createDirectStream 接收 kafka数据,这里采用是kafka低级api偏移量不受zookeeper管理
                            val dstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)
                            //7、获取kafka中topic中的数据
                    //获取元组中的值格式:元组名._下标。
                         //map(_._2):表示map遍历出的每个元素是元祖的同时,并获取元祖中下标为2的值,下标从1开始。
                         //partition.foreach(tuple=>{tuple._2} 和 map(_._2) 均都为“元组名._下标值2”,最终取出的数据才为真正要进行处理的数据。
                              val topicData: DStream[String] = dstream.map(_._2)
                            //8、切分每一行,每个单词计为1
                    //flatMap(_.split(" ")) 流的扁平化,最终输出的数据类型为一维数组Array[String],所有单词都被分割出来作为一个元素存储到同一个一维数组Array[String]
                    //map((_,1))每个单词记为1,即(单词,1),表示每个单词封装为一个元祖,其key为单词,value为1,返回MapPartitionRDD数据
                            val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_,1))
                            //9、相同单词出现的次数累加
                            //分组聚合:reduceByKey(_+_) 相同单词出现的次数累加,表示对相同的key(单词)对应的value进行累加计算 
                            val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
                            //10、打印输出
                            result.print()
                            //开启计算
                            ssc.start()
                            ssc.awaitTermination()
                        }
            }

        4.查看对应的效果
            1.向topic中添加数据,通过shell命令向topic发送消息
                cd /root/kafka
                格式:bin/kafka-console-producer.sh --broker-list kafka的IP:9092 --topic 主题名
                例子:bin/kafka-console-producer.sh --broker-list NODE1:9092 --topic kafka_spark


            2.查看控制台的输出:

 

============================================================

spark streaming从kafka获取数据,计算处理后存储到redis


因为数据要存入redis中,获取redis客户端代码如下
package com.fan.spark.stream  
   
import org.apache.commons.pool2.impl.GenericObjectPoolConfig  
import redis.clients.jedis.JedisPool  
   
/** 
  * Created by http://www.fanlegefan.com on 17-7-21. 
  */  
object RedisClient {  
  val redisHost = "127.0.0.1"  
  val redisPort = 6379  
  val redisTimeout = 30000  
   
  lazy val pool = newJedisPool(newGenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)  
  lazy val hook = newThread {  
    override def run = {  
      println("Execute hook thread: " + this)  
      pool.destroy()  
    }  
  }  
   
  sys.addShutdownHook(hook.run)  
}  


实时计算代码如下
package com.fan.spark.stream  
   
import java.text.SimpleDateFormat  
import java.util.Date  
   
import kafka.serializer.StringDecoder  
import org.apache.spark.streaming.kafka.KafkaUtils  
import org.apache.spark.streaming.{Seconds, StreamingContext}  
import org.apache.spark.{SparkConf, SparkContext}  
   
/** 
  * Created by http://www.fanlegefan.com on 17-7-21. 
  */  
object UserActionStreaming 
{     
  def main(args: Array[String]): Unit = 
  {  
    val df = newSimpleDateFormat("yyyyMMdd")  
    //设置master的地址local[N] ,n必须大于1,其中1个线程负责去接受数据,另一线程负责处理接受到的数据
    val sparkConf = newSparkConf().setAppName("pvuv").setMaster("local[3]")  
   
    val sc = newSparkContext(sparkConf)  
    //构建StreamingContext对象,每个批处理的时间间隔,即以多少秒内的数据划分为一个批次 ,当前设置 以 1 秒内的数据 划分为一个批次,每一个batch(批次)就构成一个RDD数据集
    //DStream就是一个个batch(批次)的有序序列,时间是连续的,按照时间间隔将数据流分割成一个个离散的RDD数据集
    val ssc = newStreamingContext(sc, Seconds(10))  
    ssc.checkpoint("/home/work/IdeaProjects/sparklearn/checkpoint")  
   
    val group = "test"  
    val topics = "test"  
    val topicSets = topics.split(",").toSet  //topic中的数据以","作为分隔符,即相当于把","作为回车换行的标志,进而切割出每行数据
    val kafkaParams = Map[String, String](  
      "metadata.broker.list"-> "localhost:9092",  
      "group.id"-> group  
    )  

    val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicSets)  
    //1.sparkStreaming将会创建和kafka中的topic分区数相同的rdd的分区数,而且会从kafka中并行读取数据,spark中RDD的分区数和kafka中的topic分区数是一一对应的关系。
    //2.如果kafka创建topic的命令中指定分区数为3个的话,那么就会有对应的3个RDD,所以必须使用stream.foreachRDD遍历每个RDD。
    //  同时每个RDD中又会有多个partition分区,所以必须使用rdd.foreachPartition遍历RDD中的每个partition分区。
    //  而每个partition分区中存储有topic中的数据,因此需要使用partition.foreach遍历每个partition分区中的多个元祖tuple,
    //  而每个元祖tuple封装一条数据,因此使用tuple._2 即通过“元组名._下标值2”的方式才能获取topic中真正的每行数据。
    //3.如果kafka创建topic的命令中不指定分区数的话,那么使用 stream.map(_._2) 即通过“元组名._下标值2”的方式便能获取topic中存储的所有数据。
    stream.foreachRDD(rdd=>rdd.foreachPartition(partition=>
    {  
      val jedis = RedisClient.pool.getResource  
      partition.foreach(tuple=>
      {  
        //获取元组中的值格式:元组名._下标。
        //map(_._2):表示map遍历出的每个元素是元祖的同时,并获取元祖中下标为2的值,下标从1开始。
        //partition.foreach(tuple=>{tuple._2} 和 map(_._2) 均都为“元组名._下标值2”,最终取出的数据才为真正要进行处理的数据。
        //因为使用topics.split(",").toSet,即topic中的数据把","分隔符作为回车换行的标志,所以此处获取出的数据为一行数据
        val line = tuple._2  
        //对一行数据以","分隔符进行切割,一行数据被分为4个小字符串并封装到字符串数组中
        val arr = line.split(",")  
        val user = arr(0)  //字符串数组(下标):取出一行字符串数据中的第一个被分割出来的小字符串
        val page = arr(1)  
        val money = arr(2)  
        val day = df.format(newDate(arr(3).toLong))  
        //uv  
        jedis.pfadd(day  + "_"+ page , user)  
        //pv  
        jedis.hincrBy(day+"_pv", page, 1)  
        //sum  
        jedis.hincrByFloat(day+"_sum", page, money.toDouble)  
      })  
    }))  
    ssc.start()  
    ssc.awaitTermination()  
  }  
}  

 

2019-06-30 19:37:47 ioteye 阅读数 240
  • Spark开发工程师(含项目)

    本课程为大数据金融信贷项目实战课,着重讲解企业中常用的大数据技术理论与实战,如Hadoop、Hive、HBase、Sqoop、Flume、Kafka、Spark Streaming、Spark SQL、Spark Structured Streaming等。课程包含离线项目和实时项目,从项目业务需求、技术选型、架构设计、集群安装部署、集成开发以及项目可视化进行全方位实战讲解。

    1690 人正在学习 去看看 李飞

1.0 spark原理架构

在这里插入图片描述

1.1 driver

在这里插入图片描述

1.2 cluster manager

在这里插入图片描述

1.3 DAG

在这里插入图片描述

1.4 RDD依赖关系

在这里插入图片描述

2.0 spark summary

2.1 架构图

在这里插入图片描述

2.2 项目

spark-summary

3.0 spark kafka

spark kafka

2016-08-21 22:33:52 erfucun 阅读数 3670
  • Spark开发工程师(含项目)

    本课程为大数据金融信贷项目实战课,着重讲解企业中常用的大数据技术理论与实战,如Hadoop、Hive、HBase、Sqoop、Flume、Kafka、Spark Streaming、Spark SQL、Spark Structured Streaming等。课程包含离线项目和实时项目,从项目业务需求、技术选型、架构设计、集群安装部署、集成开发以及项目可视化进行全方位实战讲解。

    1690 人正在学习 去看看 李飞

本博文讲述的内容主要包括:

1,SparkStreaming on Kafka Receiver 工作原理机制
2,SparkStreaming on Kafka Receiver案例实战
3,SparkStreaming on Kafka Receiver源码解析

一:SparkStreaming on Kafka Receiver 简介:

1、Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式,可以从代码中简单理解成Receiver方式是通过zookeeper来连接kafka队列,Direct方式是直接连接到kafka的节点上获取数据了。

2、基于Receiver的方式:

这种方式使用Receiver来获取数据。Receiver是使用Kafka的高层次Consumer API来实现的。receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的,然后Spark Streaming启动的job会去处理那些数据。
然而,在默认的配置下,这种方式可能会因为底层的失败而丢失数据。如果要启用高可靠机制,让数据零丢失,就必须启用Spark Streaming的预写日志机制(Write Ahead Log,WAL)。该机制会同步地将接收到的Kafka数据写入分布式文件系统(比如HDFS)上的预写日志中。所以,即使底层节点出现了失败,也可以使用预写日志中的数据进行恢复。

补充说明:

(1)、Kafka中的topic的partition,与Spark中的RDD的partition是没有关系的。所以,在KafkaUtils.createStream()中,提高partition的数量,只会增加一个Receiver中,读取partition的线程的数量。不会增加Spark处理数据的并行度。
(2)、可以创建多个Kafka输入DStream,使用不同的consumer group和topic,来通过多个receiver并行接收数据。
(3)、如果基于容错的文件系统,比如HDFS,启用了预写日志机制,接收到的数据都会被复制一份到预写日志中。因此,在KafkaUtils.createStream()中,设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER。

SparkStreaming on Kafka Receiver 工作原理图如下所示:

这里写图片描述

二、SparkStreaming on Kafka Receiver案例实战:

1、在进行SparkStreaming on Kafka Receiver案例的环境前提:
(1)spark 安装成功,spark 1.6.0(local方式除外)
(2)zookeeper 安装成功
(3)kafka 安装成功
(4)启动集群和zookeeper和kafka

在这里我采用local的方式进行试验,代码如下:

public class SparkStreamingOnKafkaReceiver {

    public static void main(String[] args) {
/*      第一步:配置SparkConf:
        1,至少两条线程因为Spark Streaming应用程序在运行的时候至少有一条线程用于
        不断地循环接受程序,并且至少有一条线程用于处理接受的数据(否则的话有线程用于处理数据,随着时间的推移内存和磁盘都会
        不堪重负)
        2,对于集群而言,每个Executor一般肯定不止一个线程,那对于处理SparkStreaming
        应用程序而言,每个Executor一般分配多少Core比较合适?根据我们过去的经验,5个左右的Core是最佳的(一个段子分配为奇数个Core表现最佳,例如3个,5个,7个Core等)
*/      
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("SparStreamingOnKafkaReceiver");
//      SparkConf conf = new //SparkConf().setMaster("spark://Master:7077").setAppName("        //SparStreamingOnKafkaReceiver");
/*      第二步:创建SparkStreamingContext,
        1,这个是SparkStreaming应用春香所有功能的起始点和程序调度的核心
        SparkStreamingContext的构建可以基于SparkConf参数也可以基于持久化的SparkStreamingContext的内容
//      来恢复过来(典型的场景是Driver崩溃后重新启动,由于SparkStreaming具有连续7*24
    小时不间断运行的特征,所以需要Driver重新启动后继续上一次的状态,此时的状态恢复需要基于曾经的Checkpoint))
    2,在一个Sparkstreaming 应用程序中可以创建若干个SparkStreaming对象,使用下一个SparkStreaming
        之前需要把前面正在运行的SparkStreamingContext对象关闭掉,由此,我们获取一个重大的启发
        我们获得一个重大的启发SparkStreaming也只是SparkCore上的一个应用程序而已,只不过SparkStreaming框架想运行的话需要
*/      spark工程师写业务逻辑
        @SuppressWarnings("resource")
        JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(10));

/*      第三步:创建SparkStreaming输入数据来源input Stream
        1,数据输入来源可以基于File,HDFS,Flume,Kafka-socket等
        2,在这里我们指定数据来源于网络Socket端口,SparkStreaming连接上该端口并在运行时候一直监听
        该端口的数据(当然该端口服务首先必须存在,并且在后续会根据业务需要不断地数据产生当然对于SparkStreaming
        应用程序的而言,有无数据其处理流程都是一样的);
        3,如果经常在每个5秒钟没有数据的话不断地启动空的Job其实会造成调度资源的浪费,因为并没有数据发生计算
        所以实际的企业级生成环境的代码在具体提交Job前会判断是否有数据,如果没有的话就不再提交数据
    在本案例中具体参数含义:
        第一个参数是StreamingContext实例,
        第二个参数是zookeeper集群信息(接受Kafka数据的时候会从zookeeper中获取Offset等元数据信息)
        第三个参数是Consumer Group
*/      第四个参数是消费的Topic以及并发读取Topic中Partition的线程数

        Map<String,Integer> topicConsumerConcurrency = new HashMap<String,Integer>();
        topicConsumerConcurrency.put("HelloKafakaFromSparkStreaming",1);//这里2个的话是指2个接受的线程

        JavaPairReceiverInputDStream<String, String> lines = KafkaUtils.createStream(jsc,
                "Master:2181,Worker1:2181,Worker2:2181",
                "MyFirstConsumerGrou",
                topicConsumerConcurrency);
    /*
     * 第四步:接下来就像对于RDD编程一样,基于DStream进行编程!!!原因是Dstream是RDD产生的模板(或者说类
     * ),在SparkStreaming发生计算前,其实质是把每个Batch的Dstream的操作翻译成RDD的操作
     * 对初始的DTStream进行Transformation级别处理
     * */
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<Tuple2<String, String>,String>(){ //如果是Scala,由于SAM装换,可以写成val words = lines.flatMap{line => line.split(" ")}

            @Override
            public Iterable<String> call(Tuple2<String,String> tuple) throws Exception {

                return Arrays.asList(tuple._2.split(" "));//将其变成Iterable的子类
            }
        });
//      第四步:对初始DStream进行Transformation级别操作
        //在单词拆分的基础上对每个单词进行实例计数为1,也就是word => (word ,1 )
        JavaPairDStream<String,Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word,1);
            }

        });
        //对每个单词事例技术为1的基础上对每个单词在文件中出现的总次数

         JavaPairDStream<String,Integer> wordsCount = pairs.reduceByKey(new Function2<Integer,Integer,Integer>(){

            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                // TODO Auto-generated method stub
                return v1 + v2;
            }
        });
        /*
         * 此处的print并不会直接出发Job的支持,因为现在一切都是在SparkStreaming的框架控制之下的
         * 对于spark而言具体是否触发真正的JOb运行是基于设置的Duration时间间隔的
         * 诸位一定要注意的是Spark Streaming应用程序要想执行具体的Job,对DStream就必须有output Stream操作
         * output Stream有很多类型的函数触发,类print,savaAsTextFile,scaAsHadoopFiles等
         * 其实最为重要的一个方法是foreachRDD,因为SparkStreaming处理的结果一般都会放在Redis,DB
         * DashBoard等上面,foreach主要就是用来完成这些功能的,而且可以自定义具体的数据放在哪里!!!
         * */
         wordsCount.print();

//       SparkStreaming 执行引擎也就是Driver开始运行,Driver启动的时候位于一条新线程中的,当然
//       其内部有消息接受应用程序本身或者Executor中的消息
         jsc.start();
         jsc.close();
    }


}

2、SparkStreaming on Kafka Receiver运行在集群上的步骤及结果:

1,首先启动zookeeper服务:


2,接下来启动Kafka服务


3,在eclipse上观察结果:

三:SparkStreaming on Kafka Receiver源码解析

1,首先看一下KafkaUtils(包含zookeeper的配置等等):

这里写图片描述
这里写图片描述

2、在这里创建了KafkaInputDStream:
这里写图片描述

3、这里证明KafkaInputStream为consumer
这里写图片描述

4、在这里拥有线程池(处理topic)
这里写图片描述

5,不同的接受方式(第一个为wal方式)
这里写图片描述

补充说明:

使用Spark Streaming可以处理各种数据来源类型,如:数据库、HDFS,服务器log日志、网络流,其强大超越了你想象不到的场景,只是很多时候大家不会用,其真正原因是对Spark、spark streaming本身不了解。

博文内容源自DT大数据梦工厂Spark课程。相关课程内容视频可以参考:
百度网盘链接:http://pan.baidu.com/s/1slvODe1(如果链接失效或需要后续的更多资源,请联系QQ460507491或者微信号:DT1219477246 获取上述资料)。

2018-06-09 22:33:05 m0_37803704 阅读数 1064
  • Spark开发工程师(含项目)

    本课程为大数据金融信贷项目实战课,着重讲解企业中常用的大数据技术理论与实战,如Hadoop、Hive、HBase、Sqoop、Flume、Kafka、Spark Streaming、Spark SQL、Spark Structured Streaming等。课程包含离线项目和实时项目,从项目业务需求、技术选型、架构设计、集群安装部署、集成开发以及项目可视化进行全方位实战讲解。

    1690 人正在学习 去看看 李飞

spark和kafka整合有2种方式

1、receiver

顾名思义:就是有一个线程负责获取数据,这个线程叫receiver线程

解释:

1、Spark集群中的某个executor中有一个receiver线程,这个线程负责从kafka中获取数据

 注意:这里的获取数据并不是从kafka中拉(pull) 而是接收数据,具体原理是该receiver线程发送请求到kafka,这个请求包含对kafka中每个partition的消费偏移量(offset),然后由kafka主动的推送数据到spark中,再有该receiver线程负责接收数据

2、当receiver线程接收到数据后会做备份处理,即把数据备份到其他的executor中,也可能会备份到这个receiver线程所在节点的executor中

3、当备份完毕后该线程会把每个partition的消费偏移量在zookeeper中修改,(新版本的kafka的offset 保存在kafka集群中)

4、修改完offset后,该receiver线程会把"消费"的数据告诉Driver

5、Driver分发任务时会根据每个executor上的数据,根据数据本地性发送

问题:

当第三步执行完后,对于kafka来说这一批数据已经消费完成,那么如果此时Driver挂掉,那么这一批数据就会丢失,为了解决这个问题,有一个叫WAL逾写日志的概念,即把一部分数据存储在HDFS上,当Driver回复后可以从HDFS上获取这部分数据,但是开启WAL性能会受到很大的影响

2、dirct

直接连接:即每个executor直接取kafka获取数据

1、首先Driver程序会定时(batchInterval)的向executor中发送任务(4个)

  >> 问题1:Driver怎么知道要把任务发送到哪个executor中呢?

     >> Driver会调用Kafka的接口获取某个partition位于哪个节点上,根据这个来获取这些信息并发送任务到指定的节点,这就类似于Spark集群处理HDFS上的文件数据,Spark是可以知道某些文件的block在那些节点上,就是spark调用了HDFS的相关接口

  >> 问题2:为什么是4个任务?

      >> 这个个数由消费的topic的partition的个数决定,因为spark会对每个partition开启一个任务,所以任务数是kafka的某个topic的partition数

2、当每个任务确定了处理那个partition中的数据,则就有任务本身去kafka获取数据

总结:目前公司中第二种方式使用比较多,这样也有一个问题就是说当kafka中某个topic加了ACL验证,那么这种方式是不能消费加了ACL的topic中的数据,因为kafka客户端的ACL验证需要客户端配置一个环境变量在System的Properties中,在local模式下可以实现,因为local模式下启动一个虚拟机实例,即只对应一个System,而在集群模式下,要启动多个进程,即启动多个虚拟机实例,所以System的全局属性没有办法配置

如何学习大数据?学习没有资料?

想学习大数据开发技术,Hadoop,spark,云计算,数据分析等技术,在这里向大家推荐一个学习资料分享群:894951460,里面有大牛已经整理好的相关学习资料,希望对你们有所帮助。

没有更多推荐了,返回首页