HashSet<String> topicsSet = new HashSet<String>(Arrays.asList(config.getKafkaTopics().split(",")));
HashMap<String, String> kafkaParams = new HashMap<String, String>();
kafkaParams.put("metadata.broker.list", config.getKafkaBrokers());
// Create direct KAFKA stream with brokers and topics
JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(jssc, String.class, String.class,
StringDecoder.class, StringDecoder.class, kafkaParams, topicsSet);
Tworzę strumień Kafka za pomocą createDirectStream
funkcja od KafkaUtils
, jak wyżej. Myślę, że jest to dość standardowe i myślę, że działało ze Spark-1.5.1.
Przełączyłem się na Spark-1.6.1 i chociaż nie jestem pewien, czy to z powodu wersji, powoduje to następujący błąd:
Exception in thread "main" java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.spark.deploy.worker.DriverWrapper$.main(DriverWrapper.scala:58)
at org.apache.spark.deploy.worker.DriverWrapper.main(DriverWrapper.scala)
Caused by: java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
at kafka.api.RequestKeys$.<init>(RequestKeys.scala:39)
at kafka.api.RequestKeys$.<clinit>(RequestKeys.scala)
at kafka.api.TopicMetadataRequest.<init>(TopicMetadataRequest.scala:53)
at org.apache.spark.streaming.kafka.KafkaCluster.getPartitionMetadata(KafkaCluster.scala:122)
at org.apache.spark.streaming.kafka.KafkaCluster.getPartitions(KafkaCluster.scala:112)
at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:211)
at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484)
at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:607)
at org.apache.spark.streaming.kafka.KafkaUtils.createDirectStream(KafkaUtils.scala)
at com.analytics.kafka.consumer.SystemUserAnalyticsConsumer.main(SystemUserAnalyticsConsumer.java:59)
... 6 more
Daje to bardzo mało informacji o tym, czym dokładnie jest problem.
Jaki jest tutaj problem?
Odpowiedzi:
1 dla odpowiedzi № 1Używasz niekompatybilnych wersji spark-streaming-kafka-0-10
dla wersji Scala używanych w czasie kompilacji / kompilacji i czasie wykonywania, tzn.
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.0.1"
Upewnij się, że wersje Scala są podobne (i zanotuj dwa procentowe znaki, które się nimi opiekują i polegają na nich scalaVersion
).