Spark streaming不向Cassandra插入数据

Question

线程“streaming-job-executor-53”java.lang.noClassDefounderror：com/datastax/spark/connector/columnselector在com.enerbyte.spark.jobs.wattiopipeline.wattiopipelineStreamingJob$$anonFun$main$2.在com.enerbyte.spark.jobs.wattiopipelineStreamingJob.jobs.wattiopipelineStreamingJob.wattiopipelineStreamingJob.$$anonFun$main$2$anonfun$apply$mcv$sp$1.在org.apache.spark.streaming.dstream.foreachdstream$$anonfun$1$$anonfun$apply$mcv$sp$1.在org.apache.spark.streaming.dstream.foreachdstream.scala:50(foreachdstream.scala:50)在org.apache.spark.streaming.dstream.foreachdstream.scala:50)在org.apache.spark.streaming.dstream.foreachdstream.scala:50 foreachdstream$$anonfun$1.在org.apache.spark.streaming.dstream.foreachdstream$$anonfu应用$mcv$sp（foreachdstream.scala:49)n$1.在org.apache.spark.streaming.dstream.foreachdstream.scala:49)处应用(foreachdstream.scala:49)在scala.util.try处应用(foreachdstream.scala:49)在org.apache.spark.streaming.schedule.job.run处应用(try.scala:161)在org.apache.spark.streaming.schedule.job.run处应用(job.scala:39)在.在org.apache.spark.streaming.schedule.jobscheduler.scala:224)处应用(jobscheduler.scala:224)在scala.util.dynamicvariable.With Value(dynamicvariable.scala:57)在org.apache.spark.streaming.schedule.jobschedule.scala:223)在java.util.concurrent.threadpoolexecutor.junworker（threadpoolexecutor.worker.run（：617）在java.lang.thread.run（thread.java：745）处由：java.lang.ClassNotFoundException：com.datastax.spark.conne引起ctor.columnSelector在java.net.URLClassLoader.FindClass(URLClassLoader.java:381)在java.lang.ClassLoader.LoadClass(ClassLoader.java:424)在java.lang.ClassLoader.LoadClass(ClassLoader.java:357)

我为连接器添加了依赖关系，如下所示：

“com.datastax.spark”%%“Spark-Cassandra-连接器”%“1.5.0”%“提供”

这是我的申请html" target="_blank">代码：

    val measurements = KafkaUtils.createDirectStream[
  Array[Byte],
  Array[Byte],
  DefaultDecoder,
  DefaultDecoder](ssc, kafkaConfig, Set("wattio"
))
  .map {
    case (k, v) => {
      val decoder = new AvroDecoder[WattioMeasure](null,
        WattioMeasure.SCHEMA$)
      decoder.fromBytes(v)
    }
  }

//inserting into WattioRaw
WattioFunctions.run(WattioFunctions.
  processWattioRaw(measurements))(
  (rdd: RDD[
    WattioTenantRaw], t: Time) => {
    rdd.cache()
    //get all the different tenants
    val differentTenants = rdd.map(a
    => a.tenant).distinct().collect()
    // for each tenant, create keyspace value and flush to cassandra
    differentTenants.foreach(tenant => {
      val keyspace = tenant + "_readings"
      rdd.filter(a => a.tenant == tenant).map(s => s.wattioRaw).saveToCassandra(keyspace, "wattio_raw")
    })
    rdd.unpersist(true)
  }
)

ssc.checkpoint("/tmp")
ssc.start()
ssc.awaitTermination()

Answer 1

你需要确保你的罐子可以提供给工人。一旦作业开始执行，spark master将打开一个文件服务器。

您需要使用sparkcontext.setjars或通过传递给spark-submit的--jars标志指定uber jar的路径。

从文档中