You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Sidney Feiner <si...@startapp.com> on 2017/06/01 05:54:13 UTC
RE: Message getting lost in Kafka + Spark Streaming
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.pareek@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-unsubscribe@spark.apache.org
---------------------------------------------------------------------
To unsubscribe e-mail: user-unsubscribe@spark.apache.org
Re: Message getting lost in Kafka + Spark Streaming
Posted by Vikash Pareek <vi...@infoobjects.com>.
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.pareek@infoobjects.com
On Thu, Jun 1, 2017 at 11:24 AM, Sidney Feiner <si...@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.pareek@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-unsubscribe@spark.apache.org
>
>