Spark和Kafka直接方法

我是Apache Spark的新成員，我試圖運行Spark Streaming + Kafka集成直接方法示例（JavaDirectKafkaWordCount.java）。Spark和Kafka直接方法

我已經下載了所有的圖書館，但是當我嘗試運行我得到這個錯誤

Exception in thread "main" java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object; 
at kafka.api.RequestKeys$.<init>(RequestKeys.scala:48) 
at kafka.api.RequestKeys$.<clinit>(RequestKeys.scala) 
at kafka.api.TopicMetadataRequest.<init>(TopicMetadataRequest.scala:55) 
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 it.unimi.di.luca.SimpleApp.main(SimpleApp.java:53)

有什麼建議？

來源

2016-05-04 lu_ferra

您定位的是哪個scala版本？ –

@lu_Ferra，發佈你的示例代碼可能會幫助其他人更好地回答你的任務。 – Suresh

-1

我認爲這可能是一些事情。

您可能沒有在項目中正確聲明的依賴關係。你需要確保你有kafka和火花流。如果您使用像maven這樣的構建器，則可以在此處找到需要添加到構建器文件中的行http://mvnrepository.com/
如果您嘗試讀取的主題尚不存在，那麼您也會收到錯誤消息。您可以在的東西在命令行中創建它像
```
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test 
```
請確保您有卡夫卡服務器和卡夫卡的動物園管理員運行。

如果這沒有幫助，那麼也許你應該發佈你的主。

來源

2016-05-05 02:28:26

使用下面的代碼與Scala 2.10和Kafka 0.10和Spark 1.6.2和Cassandra 3.5。

我正在使用接收器較少的方法/直接卡夫卡消費。希望有幫助

import org.apache.spark.SparkConf 
import org.apache.spark.SparkContext 
import org.apache.spark.sql.SaveMode 
import org.apache.spark.streaming.Seconds 
import org.apache.spark.streaming.StreamingContext 
import org.apache.spark.streaming.kafka.KafkaUtils 
import com.datastax.spark.connector._ 

import kafka.serializer.StringDecoder 
import org.apache.spark.rdd.RDD 
import com.datastax.spark.connector.SomeColumns 
import java.util.Formatter.DateTime 

object StreamProcessor extends Serializable { 
    def main(args: Array[String]): Unit = { 
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("StreamProcessor") 
     .set("spark.cassandra.connection.host", "127.0.0.1") 

    val sc = new SparkContext(sparkConf) 

    val ssc = new StreamingContext(sc, Seconds(2)) 

    val sqlContext = new SQLContext(sc) 

    val kafkaParams = Map("metadata.broker.list" -> "localhost:9092") 

    val topics = args.toSet 

    val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
     ssc, kafkaParams, topics) 


     stream 
    .map { 
    case (_, msg) => 
     val result = msgParseMaster(msg) 
     (result.id, result.data) 
    }.foreachRDD(rdd => if (!rdd.isEmpty)  rdd.saveToCassandra("testKS","testTable",SomeColumns("id","data"))) 

     } 
    } 

    ssc.start() 
    ssc.awaitTermination() 

    } 

    import org.json4s._ 
    import org.json4s.native.JsonMethods._ 
    case class wordCount(id: Long, data1: String, data2: String) extends serializable 
    implicit val formats = DefaultFormats 
    def msgParseMaster(msg: String): wordCount = { 
    val m = parse(msg).extract[wordCount] 
    return m 

    } 

}

來源

2016-08-08 17:58:44 Suresh

Spark和Kafka直接方法

回答

相關問題