ippontech / spark-kafka-source Goto Github PK

Hi,

First of all, thanks for your blog and code, it's been really helpful.

I am using similar logic to create stream by storing offset and sending kafka messages to elasticsearch using EsSpark.saveJsonToEs API.

However, I am facing one issue if ES cluster is down this API will try to send this message to ES for sometime and then it will throw and exception and will fetch next message from the queue, and this will go on. When ES cluster is up, we are not getting the earlier messages which was not sent as ES was down.

Could you please help me with this ?

Thanks
Akshay

kafkaStream.foreachRDD(rdd => offsetsStore.saveOffsets(rdd))
private val zkClient = new ZkClient(zkHosts, 30000, 30000,ZKStringSerializer) in Object offsetStore but the zkClient can not be Serializabled how it works?

First of all, thanks for your blog post on how to store the offsets in ZooKeeper. It would be great if you could add some license to your code. That might make it easier for other people to use it. Thanks.

Hi,
I am not sure if this is of interest to all, but it seems to be a general purpose solution. From what I saw while using this library, it does not support multiple partitions. I modified the code to support multiple partitions (and multiple topics) and am willing to submit a PR if that is desirable. Also, before submitting the PR, I wanted to make sure I have not missed out any usage pattern which provides multi partition capability. Attaching the file with the changes. Some of the references are to the files in my code base. If and when I submit the PR, I will make the required modifications
ZookeeperOffsetStore_MultiPartition.txt

Disclaimer: I have tested this for single topic with multiple partitions. Not for multiple topics.

It appears that in case repartition will be performed on the Kafka topic, from 3 partitions to 4 partitions for instance, the streaming job will ignore the new partition.
This is because providing fromOffsets parameter to createDirectStream indicates the job to read only from the 3 first partitions, and ignore the 4th one (for new records that might have been produced to it already, and will be produced during the job execution).

I would expect createDirectStream to handle non-indicated partitions (in fromOffsets) in a standard way (read new records), but unfortunately it does not behave like that.

Our application is written in Python, and we are using streaming-kafka-0-8-integration (as version 0.10 does not support python).