Thanks Sidney for your response, To check if all the messages are processed I used accumulator and also add a print statement for debuging.
*val accum = ssc.sparkContext.accumulator(0, "Debug Accumulator")* *...* *...* *...* *val mappedDataStream = dataStream.map(_._2);* * mappedDataStream.foreachRDD { rdd =>* *...* *...* *...* *partition.foreach { row =>* * if (debug) println(row.mkString)* * val keyedMessage = new KeyedMessage[String, String](props.getProperty("outTopicUnharmonized"),* * null, row.toString())* * producer.send(keyedMessage)* * println("Messges sent to Kafka: " + keyedMessage.message)* * accum += 1* * }* * //hack, should be done with the flush* * Thread.sleep(1000)* * producer.close()* * print("Accumulator's value is: " + accum)* And I am getting all the messages in "*println("Messges sent to Kafka: " + keyedMessage.message)*" received by the stream, and accumulator value is also same as number of incoming messages. Best Regards, [image: InfoObjects Inc.] <http://www.infoobjects.com/> Vikash Pareek Team Lead *InfoObjects Inc.* Big Data Analytics m: +91 8800206898 a: E5, Jhalana Institutionall Area, Jaipur, Rajasthan 302004 w: www.linkedin.com/in/pvikash e: vikash.par...@infoobjects.com On Thu, Jun 1, 2017 at 11:24 AM, Sidney Feiner <sidney.fei...@startapp.com> wrote: > Are you sure that every message gets processed? It could be that some > messages failed passing the decoder. > And during the processing, are you maybe putting the events into a map? > That way, events with the same key could override each other and that way > you'll have less final events. > > -----Original Message----- > From: Vikash Pareek [mailto:vikash.par...@infoobjects.com] > Sent: Tuesday, May 30, 2017 4:00 PM > To: user@spark.apache.org > Subject: Message getting lost in Kafka + Spark Streaming > > I am facing an issue related to spark streaming with kafka, my use case is > as > follow: > 1. Spark streaming(DirectStream) application reading data/messages from > kafka topic and process it 2. On the basis of proccessed message, app will > write proccessed message to different kafka topics for e.g. if messgese is > harmonized then write to harmonized topic else unharmonized topic > > the problem is that during the streaming somehow we are lossing some > messaged i.e all the incoming messages are not written to harmonized or > unharmonized topics. > for e.g. if app received 30 messages in one batch then sometime it write > all the messges to output topics(this is expected behaviour) but sometimes > it writes only 27 (3 messages are lost, this number can change). > > Versions as follow: > Spark 1.6.0 > Kafka 0.9 > > Kafka topics confguration is as follow: > # of brokers: 3 > # replicxation factor: 3 > # of paritions: 3 > > Following are the properties we are using for kafka: > * val props = new Properties() > props.put("metadata.broker.list", > properties.getProperty("metadataBrokerList")) > props.put("auto.offset.reset", > properties.getProperty("autoOffsetReset")) > props.put("group.id", properties.getProperty("group.id")) > props.put("serializer.class", "kafka.serializer.StringEncoder") > props.put("outTopicHarmonized", > properties.getProperty("outletKafkaTopicHarmonized")) > props.put("outTopicUnharmonized", > properties.getProperty("outletKafkaTopicUnharmonized")) > props.put("acks", "all"); > props.put("retries", "5"); > props.put("request.required.acks", "-1") > * > Following is the piece of code where we are writing proccessed messges to > kafka: > * val schemaRdd2 = finalHarmonizedDF.toJSON > > schemaRdd2.foreachPartition { partition => > val producerConfig = new ProducerConfig(props) > val producer = new Producer[String, String](producerConfig) > > partition.foreach { row => > if (debug) println(row.mkString) > val keyedMessage = new KeyedMessage[String, > String](props.getProperty("outTopicHarmonized"), > null, row.toString()) > producer.send(keyedMessage) > > } > //hack, should be done with the flush > Thread.sleep(1000) > producer.close() > } > * > We explicitely added sleep(1000) for testing purpose. > But this is also not solving the problem :( > > Any suggestion would be appreciated. > > > > -- > View this message in context: http://apache-spark-user-list. > 1001560.n3.nabble.com/Message-getting-lost-in-Kafka-Spark- > Streaming-tp28719.html > Sent from the Apache Spark User List mailing list archive at Nabble.com. > > --------------------------------------------------------------------- > To unsubscribe e-mail: user-unsubscr...@spark.apache.org > >