SparkStreaming与Kafka整合遇到的问题及解决方案

  1. 云栖社区>
  2. 博客>
  3. 正文

SparkStreaming与Kafka整合遇到的问题及解决方案

知与谁同 2017-08-03 15:42:00 浏览2723
展开阅读全文

最近工作中是做日志分析的平台,采用了sparkstreaming+kafka,采用kafka主要是看中了它对大数据量处理的高性能,处理日志类应用再好不过了,采用了sparkstreaming的流处理框架 主要是考虑到它本身是基于spark核心的,以后的批处理可以一站式服务,并且可以提供准实时服务到elasticsearch中,可以实现准实时定位系统日志。

实现

Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式。

一. 基于Receiver方式

这种方式使用Receiver来获取数据。Receiver是使用Kafka的高层次Consumer API来实现的。receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的,然后Spark Streaming启动的job会去处理那些数据。代码如下:


  1. SparkConf sparkConf = new SparkConf().setAppName("log-etl").setMaster("local[4]"); 
  2.     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000)); 
  3.     int numThreads = Integer.parseInt("4"); 
  4.     Map<String, Integer> topicMap = new HashMap<String, Integer>(); 
  5.     topicMap.put("group-45", numThreads); 
  6.      //接收的参数分别是JavaStreamingConetxt,zookeeper连接地址,groupId,kafak的topic  
  7.     JavaPairReceiverInputDStream<String, String> messages = 
  8.     KafkaUtils.createStream(jssc, "172.16.206.27:2181,172.16.206.28:2181,172.16.206.29:2181""1", topicMap); 

刚开始的时候系统正常运行,没有发现问题,但是如果系统异常重新启动sparkstreaming程序后,发现程序会重复处理已经处理过的数据,这种基于receiver的方式,是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性,但是却无法保证数据被处理一次且仅一次,可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。官方现在也已经不推荐这种整合方式,官网相关地址 http://spark.apache.org/docs/latest/streaming-kafka-integration.html ,下面我们使用官网推荐的第二种方式kafkaUtils的createDirectStream()方式。

二.基于Direct的方式

这种新的不基于Receiver的直接方式,是在Spark 1.3中引入的,从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后,这种方式会周期性地查询Kafka,来获得每个topic+partition的最新的offset,从而定义每个batch的offset的范围。当处理数据的job启动时,就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

代码如下:


  1. SparkConf sparkConf = new SparkConf().setAppName("log-etl"); 
  2. JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2)); 
  3.  
  4. HashSet<String> topicsSet = new HashSet<String>(Arrays.asList(topics.split(","))); 
  5. HashMap<String, String> kafkaParams = new HashMap<String, String>(); 
  6. kafkaParams.put("metadata.broker.list", brokers); 
  7. // Create direct kafka stream with brokers and topics 
  8. JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream( 
  9.     jssc, 
  10.     String.class, 
  11.     String.class, 
  12.     StringDecoder.class, 
  13.     StringDecoder.class, 
  14.     kafkaParams, 
  15.     topicsSet 
  16. ); 

这种direct方式的优点如下:

1.简化并行读取:如果要读取多个partition,不需要创建多个输入DStream然后对它们进行union操作。Spark会创建跟Kafka partition一样多的RDD partition,并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间,有一个一对一的映射关系。

2.一次且仅一次的事务机制:基于receiver的方式,在spark和zk中通信,很有可能导致数据的不一致。

3.高效率:在receiver的情况下,如果要保证数据的不丢失,需要开启wal机制,这种方式下,为、数据实际上被复制了两份,一份在kafka自身的副本中,另外一份要复制到wal中, direct方式下是不需要副本的。

三.基于Direct方式丢失消息的问题

貌似这种方式很完美,但是还是有问题的,当业务需要重启sparkstreaming程序的时候,业务日志依然会打入到kafka中,当job重启后只能从最新的offset开始消费消息,造成重启过程中的消息丢失。kafka中的offset如下图(使用kafkaManager实时监控队列中的消息):

当停止业务日志的接受后,先重启spark程序,但是发现job并没有将先前打入到kafka中的数据消费掉。这是因为消息没有经过zk,topic的offset也就没有保存

四.解决消息丢失的处理方案

一般有两种方式处理这种问题,可以先spark streaming 保存offset,使用spark checkpoint机制,第二种是程序中自己实现保存offset逻辑,我比较喜欢第二种方式,以为这种方式可控,所有主动权都在自己手中。

先看下大体流程图,


  1. SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("log-etl"); 
  2.  Set<String> topicSet = new HashSet<String>(); 
  3.         topicSet.add("group-45"); 
  4.         kafkaParam.put("metadata.broker.list""172.16.206.17:9092,172.16.206.31:9092,172.16.206.32:9092"); 
  5.         kafkaParam.put("group.id""simple1"); 
  6.  
  7.         // transform java Map to scala immutable.map 
  8.         scala.collection.mutable.Map<String, String> testMap = JavaConversions.mapAsScalaMap(kafkaParam); 
  9.         scala.collection.immutable.Map<String, String> scalaKafkaParam = 
  10.                 testMap.toMap(new Predef.$less$colon$less<Tuple2<String, String>, Tuple2<String, String>>() { 
  11.                     public Tuple2<String, String> apply(Tuple2<String, String> v1) { 
  12.                         return v1; 
  13.                     } 
  14.                 }); 
  15.  
  16.         // init KafkaCluster 
  17.         kafkaCluster = new KafkaCluster(scalaKafkaParam); 
  18.  
  19.         scala.collection.mutable.Set<String> mutableTopics = JavaConversions.asScalaSet(topicSet); 
  20.         immutableTopics = mutableTopics.toSet(); 
  21.         scala.collection.immutable.Set<TopicAndPartition> topicAndPartitionSet2 = kafkaCluster.getPartitions(immutableTopics).right().get(); 
  22.  
  23.         // kafka direct stream 初始化时使用的offset数据 
  24.         Map<TopicAndPartition, Long> consumerOffsetsLong = new HashMap<TopicAndPartition, Long>(); 
  25.  
  26.         // 没有保存offset时(该group首次消费时), 各个partition offset 默认为0 
  27.         if (kafkaCluster.getConsumerOffsets(kafkaParam.get("group.id"), topicAndPartitionSet2).isLeft()) { 
  28.  
  29.             System.out.println(kafkaCluster.getConsumerOffsets(kafkaParam.get("group.id"), topicAndPartitionSet2).left().get()); 
  30.  
  31.             Set<TopicAndPartition> topicAndPartitionSet1 = JavaConversions.setAsJavaSet((scala.collection.immutable.Set)topicAndPartitionSet2); 
  32.  
  33.             for (TopicAndPartition topicAndPartition : topicAndPartitionSet1) { 
  34.                 consumerOffsetsLong.put(topicAndPartition, 0L); 
  35.             } 
  36.  
  37.         } 
  38.         // offset已存在, 使用保存的offset 
  39.         else { 
  40.  
  41.             scala.collection.immutable.Map<TopicAndPartition, Object> consumerOffsetsTemp = kafkaCluster.getConsumerOffsets("simple1", topicAndPartitionSet2).right().get(); 
  42.  
  43.             Map<TopicAndPartition, Object> consumerOffsets = JavaConversions.mapAsJavaMap((scala.collection.immutable.Map)consumerOffsetsTemp); 
  44.  
  45.             Set<TopicAndPartition> topicAndPartitionSet1 = JavaConversions.setAsJavaSet((scala.collection.immutable.Set)topicAndPartitionSet2); 
  46.  
  47.             for (TopicAndPartition topicAndPartition : topicAndPartitionSet1) { 
  48.                 Long offset = (Long)consumerOffsets.get(topicAndPartition); 
  49.                 consumerOffsetsLong.put(topicAndPartition, offset); 
  50.             } 
  51.  
  52.         } 
  53.  
  54.         JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(5000)); 
  55.         kafkaParamBroadcast = jssc.sparkContext().broadcast(kafkaParam); 
  56.  
  57.         // create direct stream 
  58.         JavaInputDStream<String> message = KafkaUtils.createDirectStream( 
  59.                 jssc, 
  60.                 String.class, 
  61.                 String.class, 
  62.                 StringDecoder.class, 
  63.                 StringDecoder.class, 
  64.                 String.class, 
  65.                 kafkaParam, 
  66.                 consumerOffsetsLong, 
  67.                 new Function<MessageAndMetadata<String, String>, String>() { 
  68.                     public String call(MessageAndMetadata<String, String> v1) throws Exception { 
  69.                         System.out.println("接收到的数据《《==="+v1.message()); 
  70.                         return v1.message(); 
  71.                     } 
  72.                 } 
  73.         ); 
  74.  
  75.         // 得到rdd各个分区对应的offset, 并保存在offsetRanges中 
  76.         final AtomicReference<OffsetRange[]> offsetRanges = new AtomicReference<OffsetRange[]>(); 
  77.  
  78.         JavaDStream<String> javaDStream = message.transform(new Function<JavaRDD<String>, JavaRDD<String>>() { 
  79.             public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception { 
  80.                 OffsetRange[] offsets = ((HasOffsetRanges) rdd.rdd()).offsetRanges(); 
  81.                 offsetRanges.set(offsets); 
  82.                 return rdd; 
  83.             } 
  84.         }); 
  85.  
  86.         // output 
  87.         javaDStream.foreachRDD(new Function<JavaRDD<String>, Void>() { 
  88.  
  89.             public Void call(JavaRDD<String> v1) throws Exception { 
  90.                 if (v1.isEmpty()) return null
  91.  
  92.                 List<String> list = v1.collect(); 
  93.                 for(String s:list){ 
  94.                     System.out.println("数据==="+s); 
  95.                 } 
  96.  
  97.                 for (OffsetRange o : offsetRanges.get()) { 
  98.  
  99.                     // 封装topic.partition 与 offset对应关系 java Map 
  100.                     TopicAndPartition topicAndPartition = new TopicAndPartition(o.topic(), o.partition()); 
  101.                     Map<TopicAndPartition, Object> topicAndPartitionObjectMap = new HashMap<TopicAndPartition, Object>(); 
  102.                     topicAndPartitionObjectMap.put(topicAndPartition, o.untilOffset()); 
  103.  
  104.                     // 转换java map to scala immutable.map 
  105.                     scala.collection.mutable.Map<TopicAndPartition, Object> testMap = 
  106.                             JavaConversions.mapAsScalaMap(topicAndPartitionObjectMap); 
  107.                     scala.collection.immutable.Map<TopicAndPartition, Object> scalatopicAndPartitionObjectMap = 
  108.                             testMap.toMap(new Predef.$less$colon$less<Tuple2<TopicAndPartition, Object>, Tuple2<TopicAndPartition, Object>>() { 
  109.                                 public Tuple2<TopicAndPartition, Object> apply(Tuple2<TopicAndPartition, Object> v1) { 
  110.                                     return v1; 
  111.                                 } 
  112.                             }); 
  113.  
  114.                     // 更新offset到kafkaCluster 
  115.                     kafkaCluster.setConsumerOffsets(kafkaParamBroadcast.getValue().get("group.id"), scalatopicAndPartitionObjectMap); 
  116.                        System.out.println("原数据====》"+o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset() 
  117.                     ); 
  118.                 } 
  119.                 return null
  120.             } 
  121.         }); 
  122.  
  123.         jssc.start(); 
  124.         jssc.awaitTermination(); 
  125.     } 

基本使用这种方式就可以解决数据丢失的问题。


本文作者:佚名

来源:51CTO

网友评论

登录后评论
0/500
评论
知与谁同
+ 关注