8620-84511745

Blog Category

我们的最新消息和博客

Spark Streaming kafka 实现数据零丢失的几种方式

Posted in 大数据 on Jul 27, 2016

定义

问题开始之前先解释下流处理中的一些概念:

  • At most once - 每条数据最多被处理一次(0次或1次)
  • At least once - 每条数据最少被处理一次 (1次或更多)
  • Exactly once - 每条数据只会被处理一次(没有数据会丢失,并且没有数据会被多次处理)
High Level API

如果不做容错,将会带来数据丢失

因为receiver一直在接收数据,在其没有处理的时候(已通知zk数据接收到),executor突然挂掉(或是driver挂掉通知executor关闭),缓存在其中的数据就会丢失。

因为这个问题,Spark1.2开始加入了WAL(Write ahead log

开启 WAL,将receiver获取数据的存储级别修改为StorageLevel.MEMORY_AND_DISK_SER

val conf = new SparkConf()

conf.set("spark.streaming.receiver.writeAheadLog.enable","true")

val sc= new SparkContext(conf)

val ssc = new StreamingContext(sc,Seconds(5))

ssc.checkpoint("walDir")

val lines = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicMap, StorageLevel.MEMORY_AND_DISKSER).map(._2)

开启WAL后,依旧存在数据丢失问题

即使按官方说的设置了WAL,依旧会有数据丢失,这是为什么?因为在任务中断时receiver也被强行终止了,将会造成数据丢失,提示如下:

ERROR ReceiverTracker: Deregistered receiver for stream 0: Stopped by driver

WARN BlockGenerator: Cannot stop BlockGenerator as its not in the Active state [state = StoppedAll]

WARN BatchedWriteAheadLog: BatchedWriteAheadLog Writer queue interrupted.

在Streaming程序的最后添加代码,只有在确认所有receiver都关闭的情况下才终止程序。

sys.addShutdownHook({ ssc.stop(true,true)})

调用的方法为:

def stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit

WAL带来的问题

WAL实现的是At-least-once语义。 如果在写入到外部存储的数据还没有将offset更新到zookeeper就挂掉,这些数据将会被反复消费。同时,降低了程序的吞吐量。

Kafka Direct API

Kafka direct API 的运行方式,将不再使用receiver来读取数据,也不用使用WAL机制。

同时保证了exactly-once语义,不会在WAL中消费重复数据。不过需要自己完成将offset写入zk的过程,在官方文档中都有相应介绍。

例如如下的调用方式:

messages.foreachRDD(rdd=>{ val message = rdd.map(_._2) //对数据进行一些操作

message.map(method)//更新zk上的offset (自己实现)

updateZKOffsets(rdd) })

原文链接:http://www.jianshu.com/p/716af5449175

首席工程师揭秘:LinkedIn大数据后台是如何运作的

Posted in 大数据 on May 23, 2016

“不懂得日志,你就不可能完全懂得数据库”Jay Kreps说道,Jay Kreps是LinkedIn公司首席工程师,本文介绍他本人对于日志的心得体会,包括日志是什么,如何在数据集成、实时处理和系统构建中使用日志等