zoukankan      html  css  js  c++  java
  • spark streaming集成kafka接收数据的方式

    spark streaming是以batch的方式来消费,strom是准实时一条一条的消费。当然也可以使用trident和tick的方式来实现batch消费(官方叫做mini batch)。效率嘛,有待验证。不过这两种方式都是先把数据从kafka中读取出来,然后缓存在内存或者第三方,再定时处理。如果这时候集群退出,而偏移量又没处理好的话,数据就丢掉了。

    而spark streaming提供了两种获取方式,一种是同storm一样,实时读取缓存到内存中;另一种是定时批量读取。

    这两种方式分别是:

      Receiver-base

      Direct

     


    下面分别介绍两种方式的实现

    Receiver-base

    spark streaming启动过后,会选择一台excetor作为ReceiverSupervior

    1:Reciver的父级ReciverTracker分发多个job(task)到不同的executor,并启动ReciverSupervisor.

    2:ReceiverSupervior会启动对应的实例reciver(kafkareciver,TwitterReceiver),并调用onstart()

    3:kafkareciver在通过onstart()启动后就开启线程源源不断的接收数据,并交给ReceiverSupervior,通过ReceiverSupervior.store函数一条一条接收

    4:ReceiverSupervior会调用BlockGenertor.adddata填充数据。

    所有的中间数据都缓存在BlockGenertor

    1:首先BlockGenertor维护了一个缓冲区,currentbuffer,一个无限长度的arraybuffer。为了防止内存撑爆,这个currentbuffer的大小可以被限制,通过设置参数spark.streaming.reciver.maxRate,以秒为单位。currentbuffer所使用的内存不是storage(负责spark计算过程中的所有存储,包括磁盘和内存),而是珍贵的计算内存。所以currentbuffer应该被限制,防止占用过多计算内存,拖慢任务计算效率,甚至有可能拖垮Executor甚至集群。

    2:维护blockforpushing队列,它是等待被拉到到BlockManager的中转站。它是currentbuffer和BlockManager的中间环节。它里面的每一个元素其实就是一个currentbuffer。

    3:维护两个定时器,其实就是一个生产-消费模式。blockintervaltimer定时器,负责生产端,定时将currentbuffer放进blockforpushing队列。blockforpushingthread负责消费端,定时将blockforpushing里的数据转移到BlockManager。

    Direct

    首先这种方式是延迟的。也就是说当action真正触发时才会去kafka里接数据。因此不存在currentbuffer的概念。它把kafka每个分区里的数据,映射为KafkaRdd的概念。题外话,在structured streaming中,也已经向DataFrame和DataSet统一了,弱化了RDD的概念。

    真正与kafka打交道的是KafkaCluster,全限定名: org.apache.spark.streaming.kafka.KafkaCluster。包括设备kafka各种参数,连接,获取分区,以及偏移量,设置偏移量范围等。

  • 相关阅读:
    树链剖分 (模板) 洛谷3384
    ST表 (模板) 洛谷3865
    IOI 2005 River (洛谷 3354)
    IOI 2005 River (洛谷 3354)
    poj1094 Sorting It All Out
    poj1094 Sorting It All Out
    spfa(模板)
    HAOI 2006 受欢迎的牛 (洛谷2341)
    HAOI 2006 受欢迎的牛 (洛谷2341)
    洛谷1850(NOIp2016) 换教室——期望dp
  • 原文地址:https://www.cnblogs.com/eryuan/p/7065242.html
Copyright © 2011-2022 走看看