Spark Streaming 'numRecords must not be negative'问题解决

转载自：http://blog.csdn.net/xueba207/article/details/51135423

问题描述

笔者使用spark streaming读取Kakfa中的数据，做进一步处理，用到了KafkaUtil的createDirectStream()方法；该方法不会自动保存topic partition的offset到zk，需要在代码中编写提交逻辑，此处介绍了保存offset的方法。 
当删除已经使用过的kafka topic，然后新建同名topic，使用该方式时出现了"numRecords must not be negative"异常 
详细信息如下图: 

是不合法的参数异常，RDD的记录数目必须不能是负数。 
下文详细分析该问题的出现的场景，以及解决方法。

异常分析

numRecords确定

首先，定位出异常出现的问题，和大致原因。异常中打印出了出现的位置 org.apache.spark.streaming.scheduler.StreamInputInfo.InputInfoTracker的第38行，此处代码：

代码38行，判断了numRecords是否大于等于0，当不满足条件时抛出异常，可判断此时numRecords<0。 
numRecords的解释： 
numRecords: the number of records in a batch 
应该是当前rdd中records 数目计算出了问题。 
numRecords 构造StreamInputInfo时的参数，结合异常中的信息，找到了DirectKafkaInputDStream中的构造InputInfo的位置： 

可知 numRecords是rdd.count()的值。

rdd.count的计算

根据以上分析可知rdd.count()值为负值，因此需要分析rdd的是如何生成的。 
同样在DirectKafkaInputDStream中找到rdd的生成代码：

从此处一路跟踪代码，可在KafkaRDD.scala中找到rdd.count的赋值逻辑：

offsetRanges的计算逻辑

offsetRanges的定义

offsetRanges: offset ranges that define the Kafka data belonging to this RDD

在KafkaRDDPartition 40行找到kafka partition offsetRange的计算逻辑：

def count(): Long = untilOffset - fromOffset 
fromOffset： per-topic/partition Kafka offset defining the (inclusive) starting point of the batch 
untilOffset： per-topic/partition Kafka offset defining the (inclusive) ending point of the batch

fromOffset来自zk中保存； 
untilOffset通过DirectKafkaInputDStream第145行：

val untilOffsets = clamp(latestLeaderOffsets(maxRetries))

计算得到，计算过程得到最新的offset，然后使用spark.streaming.kafka.maxRatePerPartition做clamp，得到允许的最大untilOffsets，##而此时新建的topic，如果topic中没有数据，untilOffsets应该为0##

原因总结

当删除一个topic时，zk中的offset信息并没有被清除，因此KafkaDirectStreaming再次启动时仍会得到旧的topic offset为old_offset，作为fromOffset。 
当新建了topic后，使用untiloffset计算逻辑，得到untilOffset为0（如果topic已有数据则>0）； 
再次被启动的KafkaDirectStreaming Job通过异常的计算逻辑得到的rdd numRecords值为可计算为： 
numRecords = untilOffset - fromOffset(old_offset) 
当untilOffset < old_offset时，此异常会出现，对于新建的topic这种情况的可能性很大

解决方法

思路

根据以上分析，可在确定KafkaDirectStreaming 的fromOffsets时判断fromOffset与untiloffset的大小关系，当untilOffset < fromOffset时，矫正fromOffset为offset初始值0。

流程

• 从zk获取topic/partition 的fromOffset（获取方法链接）
• 利用SimpleConsumer获取每个partiton的lastOffset（untilOffset ）
• 判断每个partition lastOffset与fromOffset的关系
• 当lastOffset < fromOffset时，将fromOffset赋值为0 
  通过以上步骤完成fromOffset的值矫正。

核心代码

获取kafka topic partition lastoffset代码：

1. package org.frey.example.utils.kafka;
2.  
3. import com.google.common.collect.Lists;
4. import com.google.common.collect.Maps;
5. import kafka.api.PartitionOffsetRequestInfo;
6. import kafka.cluster.Broker;
7. import kafka.common.TopicAndPartition;
8. import kafka.javaapi.*;
9. import kafka.javaapi.consumer.SimpleConsumer;
10.  
11. import java.util.Date;
12. import java.util.HashMap;
13. import java.util.List;
14. import java.util.Map;
15.  
16. /**
17.  * KafkaOffsetTool
18.  *
19.  * @author v1-daddy
20.  * @date 2016/4/11
21.  */
22. public class KafkaOffsetTool {
23.  
24.   private static KafkaOffsetTool instance;
25.   final int TIMEOUT = 100000;
26.   final int BUFFERSIZE = 64 * 1024;
27.  
28.   private KafkaOffsetTool() {
29.   }
30.  
31.   public static synchronized KafkaOffsetTool getInstance() {
32.     if (instance == null) {
33.       instance = new KafkaOffsetTool();
34.     }
35.     return instance;
36.   }
37.  
38.   public Map<TopicAndPartition, Long> getLastOffset(String brokerList, List<String> topics,
39.       String groupId) {
40.  
41.     Map<TopicAndPartition, Long> topicAndPartitionLongMap = Maps.newHashMap();
42.  
43.     Map<TopicAndPartition, Broker> topicAndPartitionBrokerMap =
44.         KafkaOffsetTool.getInstance().findLeader(brokerList, topics);
45.  
46.     for (Map.Entry<TopicAndPartition, Broker> topicAndPartitionBrokerEntry : topicAndPartitionBrokerMap
47.         .entrySet()) {
48.       // get leader broker
49.       Broker leaderBroker = topicAndPartitionBrokerEntry.getValue();
50.  
51.       SimpleConsumer simpleConsumer = new SimpleConsumer(leaderBroker.host(), leaderBroker.port(),
52.           TIMEOUT, BUFFERSIZE, groupId);
53.  
54.       long readOffset = getTopicAndPartitionLastOffset(simpleConsumer,
55.           topicAndPartitionBrokerEntry.getKey(), groupId);
56.  
57.       topicAndPartitionLongMap.put(topicAndPartitionBrokerEntry.getKey(), readOffset);
58.  
59.     }
60.  
61.     return topicAndPartitionLongMap;
62.  
63.   }
64.  
65.   /**
66.    * 得到所有的 TopicAndPartition
67.    *
68.    * @param brokerList
69.    * @param topics
70.    * @return topicAndPartitions
71.    */
72.   private Map<TopicAndPartition, Broker> findLeader(String brokerList, List<String> topics) {
73.     // get broker's url array
74.     String[] brokerUrlArray = getBorkerUrlFromBrokerList(brokerList);
75.     // get broker's port map
76.     Map<String, Integer> brokerPortMap = getPortFromBrokerList(brokerList);
77.  
78.     // create array list of TopicAndPartition
79.     Map<TopicAndPartition, Broker> topicAndPartitionBrokerMap = Maps.newHashMap();
80.  
81.     for (String broker : brokerUrlArray) {
82.  
83.       SimpleConsumer consumer = null;
84.       try {
85.         // new instance of simple Consumer
86.         consumer = new SimpleConsumer(broker, brokerPortMap.get(broker), TIMEOUT, BUFFERSIZE,
87.             "leaderLookup" + new Date().getTime());
88.  
89.         TopicMetadataRequest req = new TopicMetadataRequest(topics);
90.  
91.         TopicMetadataResponse resp = consumer.send(req);
92.  
93.         List<TopicMetadata> metaData = resp.topicsMetadata();
94.  
95.         for (TopicMetadata item : metaData) {
96.           for (PartitionMetadata part : item.partitionsMetadata()) {
97.             TopicAndPartition topicAndPartition =
98.                 new TopicAndPartition(item.topic(), part.partitionId());
99.             topicAndPartitionBrokerMap.put(topicAndPartition, part.leader());
100.           }
101.         }
102.       } catch (Exception e) {
103.         e.printStackTrace();
104.       } finally {
105.         if (consumer != null)
106.           consumer.close();
107.       }
108.     }
109.     return topicAndPartitionBrokerMap;
110.   }
111.  
112.   /**
113.    * get last offset
114.    * @param consumer
115.    * @param topicAndPartition
116.    * @param clientName
117.    * @return
118.    */
119.   private long getTopicAndPartitionLastOffset(SimpleConsumer consumer,
120.       TopicAndPartition topicAndPartition, String clientName) {
121.     Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
122.         new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
123.  
124.     requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(
125.         kafka.api.OffsetRequest.LatestTime(), 1));
126.  
127.     OffsetRequest request = new OffsetRequest(
128.         requestInfo, kafka.api.OffsetRequest.CurrentVersion(),
129.         clientName);
130.  
131.     OffsetResponse response = consumer.getOffsetsBefore(request);
132.  
133.     if (response.hasError()) {
134.       System.out
135.           .println("Error fetching data Offset Data the Broker. Reason: "
136.               + response.errorCode(topicAndPartition.topic(), topicAndPartition.partition()));
137.       return 0;
138.     }
139.     long[] offsets = response.offsets(topicAndPartition.topic(), topicAndPartition.partition());
140.     return offsets[0];
141.   }
142.   /**
143.    * 得到所有的broker url
144.    *
145.    * @param brokerlist
146.    * @return
147.    */
148.   private String[] getBorkerUrlFromBrokerList(String brokerlist) {
149.     String[] brokers = brokerlist.split(",");
150.     for (int i = 0; i < brokers.length; i++) {
151.       brokers[i] = brokers[i].split(":")[0];
152.     }
153.     return brokers;
154.   }
155.  
156.   /**
157.    * 得到broker url 与 其port 的映射关系
158.    *
159.    * @param brokerlist
160.    * @return
161.    */
162.   private Map<String, Integer> getPortFromBrokerList(String brokerlist) {
163.     Map<String, Integer> map = new HashMap<String, Integer>();
164.     String[] brokers = brokerlist.split(",");
165.     for (String item : brokers) {
166.       String[] itemArr = item.split(":");
167.       if (itemArr.length > 1) {
168.         map.put(itemArr[0], Integer.parseInt(itemArr[1]));
169.       }
170.     }
171.     return map;
172.   }
173.  
174.   public static void main(String[] args) {
175.     List<String> topics = Lists.newArrayList();
176.     topics.add("ys");
177.     topics.add("bugfix");
178.     Map<TopicAndPartition, Long> topicAndPartitionLongMap =
179.         KafkaOffsetTool.getInstance().getLastOffset("broker001:9092,broker002:9092", topics, "my.group.id");
180.  
181.     for (Map.Entry<TopicAndPartition, Long> entry : topicAndPartitionLongMap.entrySet()) {
182.      System.out.println(entry.getKey().topic() + "-"+ entry.getKey().partition() + ":" + entry.getValue());
183.     }
184.   }
185. }

 

矫正offset核心代码：

1.     /** 以下 矫正 offset */
2.     // 得到Topic/partition 的lastOffsets
3.     Map<TopicAndPartition, Long> topicAndPartitionLongMap =
4.         KafkaOffsetTool.getInstance().getLastOffset(kafkaParams.get("metadata.broker.list"),
5.             topicList, "my.group.id");
6.  
7.     // 遍历每个Topic.partition
8.     for (Map.Entry<TopicAndPartition, Long> topicAndPartitionLongEntry : fromOffsets.entrySet()) {
9.       // fromOffset > lastOffset时
10.       if (topicAndPartitionLongEntry.getValue() >
11.           topicAndPartitionLongMap.get(topicAndPartitionLongEntry.getKey())) {
12.          //矫正fromoffset为offset初始值0
13.         topicAndPartitionLongEntry.setValue(0L);
14.       }
15.     }
16.     /** 以上 矫正 offset */​