SparkStreaming+kafka Receiver模式

1.图解

2.过程

　　1.使用Kafka的High Level Consumer API 实现，消费者不能自己去维护消费者offset，而且kafka也不关心数据是否丢失。

　　2.当向zookeeper中更新完offset后，Driver如果挂到，Driver下的Executors会被kill掉，会造成数据丢失。

　　3.开启WAL【Write Ahead Log】预写日志机制，将数据备份到HDFS中一份，再去更新zookeeper中的offset，此时需调整spark存储基本，去掉备份两次【MEMORY_AND_DISK_SER_2中的_2】。开启WAL机制会加大application处理的时间。

3.特点

　　1.receiver模式依赖zookeeper管理offset。

　　2.receiver模式的并行度由spark.streaming.blockInterval决定，默认是200ms。

　　3.receiver模式接收block.batch数据后会封装到RDD中，这里的block对应RDD中的partition。

　　4.在batchInterval一定的情况下，减少spark.streaming.Interval参数值，会增大DStream中的partition个数，建议spark.streaming.Interval最低不能低于50ms。

4.代码实现

package big.data.analyse.streaming

import org.apache.spark.storage.StorageLevel

import org.apache.spark.streaming.kafka.KafkaUtils

import org.apache.spark.streaming.{Seconds, StreamingContext}

import org.apache.spark.{SparkContext, SparkConf}

/**

  * Created by zhen on 2019/5/11.

  */

object SparkStreamingReceiverKafka {

  def main(args: Array[String]) {

    val conf = new SparkConf()

    conf.setAppName("SparkStreamingReceiverKafka")

    conf.set("spark.streaming.kafka.maxRatePerPartition", "")

    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)

    sc.setLogLevel("WARN")

    val ssc = new StreamingContext(sc, Seconds()) // 创建streamingcontext入口

    val quorum = "master,worker1,worker2"

    val groupId = "zhenGroup"

    val map : Map[String, Int] = Map("zhenTopic" -> ) // topic名称为zhenTopic，每次使用1个线程读取数据

    val dframe = KafkaUtils.createStream(ssc, quorum, groupId, map, StorageLevel.MEMORY_AND_DISK_SER_2)

    dframe.foreachRDD(rdd => { // 操作方式和rdd类似，必须使用action算子才会触发程序执行！

      rdd.foreachPartition(partition =>{

        partition.foreach(println)

      })

    })

  }

}

巴特西

SparkStreaming+kafka Receiver模式

1.图解

2.过程

3.特点

4.代码实现

最新文章

热门文章