Hi Saisai, thanks you for your help, all is working ok now.
Cheers!
2014-10-09 2:49 GMT+02:00 Shao, Saisai saisai.s...@intel.com:
Hi, I think you have to change the code like this to specify the type
info, like this:
* val kafkaStream: ReceiverInputDStream[(String, String)] =
KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](*
*ssc,*
*kafkaParams,*
*topicMap, StorageLevel.MEMORY_AND_DISK_SER_2)*
You can take a try, actually Kafka unit test also use this API and worked
fine.
Besides, the fixed issue you mentioned below will only be occurred in Java
code calling related API.
Thanks
Jerry
*From:* Antonio Jesus Navarro [mailto:ajnava...@stratio.com]
*Sent:* Wednesday, October 08, 2014 10:04 PM
*To:* user@spark.apache.org
*Subject:* Error reading from Kafka
Hi, I'm trying to read from Kafka. I was able to do it correctly with
this method.
def createStream(
ssc: StreamingContext,
zkQuorum: String,
groupId: String,
topics: Map[String, Int],
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[(String, String)]
But now I have to add some params to kafka consumer so I've changed to
other createStream method but I'm getting an error:
14/10/08 15:34:10 INFO receiver.ReceiverSupervisorImpl: Deregistering
receiver 0
*14/10/08 15:34:10 ERROR scheduler.ReceiverTracker: Deregistered receiver
for stream 0: Error starting receiver 0 - java.lang.NoSuchMethodException:
scala.runtime.Nothing$.init(kafka.utils.VerifiableProperties)*
at java.lang.Class.getConstructor0(Class.java:2849)
at java.lang.Class.getConstructor(Class.java:1718)
at
org.apache.spark.streaming.kafka.KafkaReceiver.onStart(KafkaInputDStream.scala:106)
at
org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:121)
at
org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:106)
at
org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:264)
at
org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverLauncher$$anonfun$9.apply(ReceiverTracker.scala:257)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1121)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
at org.apache.spark.scheduler.Task.run(Task.scala:54)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
*This is my code. It seems that createStream returns
ReceiverInputDStream[(Nothing, Nothing)] (forced by me to (string, string))
so, I think that try togetConstructor(kafka.utils.VerifiableProperties) by
reflection from Nothing object and don't find the method. *
*val topics = config.getString(nessus.kafka.topics)*
* val numThreads = config.getInt(nessus.kafka.numThreads)*
* val topicMap = topics.split(,).map((_,numThreads.toInt)).toMap*
* val kafkaParams = Map(*
*zookeeper.connect - localhost:2181,*
*group.id http://group.id/ - my-grp)*
* val kafkaStream: ReceiverInputDStream[(String, String)] =
KafkaUtils.createStream(ssc,*
*kafkaParams,*
*topicMap, StorageLevel.MEMORY_AND_DISK_SER_2)*
I found that issue https://issues.apache.org/jira/browse/SPARK-2103
https://issues.apache.org/jira/browse/SPARK-2103 but it was solved and
I'm using spark 1.1.0 and scala 2.10 so I don't know what happens.
Any thoughts?
--
http://www.stratio.com/
Avenida de Europa, 26. Ática 5. 3ª Planta
28224 Pozuelo de Alarcón, Madrid
Tel: +34 91 352 59 42 // *@stratiobd https://twitter.com/StratioBD*
--
http://www.stratio.com/
Avenida de Europa, 26. Ática 5. 3ª Planta
28224 Pozuelo de Alarcón, Madrid
Tel: +34 91 352 59 42 // *@stratiobd https://twitter.com/StratioBD*
