我用Pypark和Kafka接收器来处理推特流。我的应用程序的一个步骤包括调用google自然语言api来获得每条tweet的情绪得分。然而，我看到api在每个处理过的tweet上都会收到几个调用（我在google云控制台中看到了调用的数量）。
另外，如果我打印tweetids（在Map函数中），我会得到3到4次相同的id。在我的应用程序结束时，tweets被发送到kafka的另一个主题，在那里我得到了正确的tweets计数（没有重复的id），所以原则上一切都正常工作，但我不知道如何避免每次tweet调用google api不止一次。
这与spark或kafka中的某些配置参数有关吗？
下面是我的控制台输出示例：

TIME 21:53:36: Google Response for tweet 801181843500466177 DONE!
TIME 21:53:36: Google Response for tweet 801181854766399489 DONE!
TIME 21:53:36: Google Response for tweet 801181844808966144 DONE!
TIME 21:53:37: Google Response for tweet 801181854372012032 DONE!
TIME 21:53:37: Google Response for tweet 801181843500466177 DONE!
TIME 21:53:37: Google Response for tweet 801181854766399489 DONE!
TIME 21:53:37: Google Response for tweet 801181844808966144 DONE!
TIME 21:53:37: Google Response for tweet 801181854372012032 DONE!

但是在Kafka接收器中，我只收到4条经过处理的tweet（这是正确的接收方式，因为它们只有4条独特的tweet）。
执行此操作的代码是：

def sendToKafka(rdd,topic,address):
    publish_producer = KafkaProducer(bootstrap_servers=address,\
                            value_serializer=lambda v: json.dumps(v).encode('utf-8'))
    records = rdd.collect()
    msg_dict = defaultdict(list)
    for rec in records:
        msg_dict["results"].append(rec)
    publish_producer.send(resultTopic,msg_dict)
    publish_producer.close()
kafka_stream = KafkaUtils.createStream(ssc, zookeeperAddress, "spark-consumer-"+myTopic, {myTopic: 1})
dstream_tweets=kafka_stream.map(lambda kafka_rec: get_json(kafka_rec[1]))\
                 .map(lambda post: add_normalized_text(post))\
                 .map(lambda post: tagKeywords(post,tokenizer,desired_keywords))\
                 .filter(lambda post: post["keywords"] == True)\
                 .map(lambda post: googleNLP.complementTweetFeatures(post,job_id))
dstream_tweets.foreachRDD(lambda rdd: sendToKafka(rdd,resultTopic,PRODUCER_ADDRESS))