如何将spark与多个cassandra主机的访问和不同的表进行集成，以处理不同的方法

我想把spark和cassandra结合起来。我有多种方法，需要访问不同的cassandra主机和不同的表。我找不到该怎么处理。请帮我解决这个问题。
以下是程序代码：

def insert(data: TripHistoryData) {

      var em=sc.parallelize(Seq(data))
             em.saveToCassandra("ap", "trip_summary_data",SomeColumns( "service_id" ,"asset_id","summ_typ","summ_dt","trp_summ_id","asset_serial_no","avg_sp","c_dist","c_epa","c_gal","c_mil","device_id","device_serial_no","dist","en_addr","en_dt","en_lat","en_long","epa","gal","h_dist","h_epa","h_gal","h_mil","id_tm","max_sp","mil","rec_crt_dt","st_addr","st_lat","st_long","tr_dis","tr_dt","tr_dur"))

     }

 def update(data: TripHistoryData) {

        var em=sc.parallelize(Seq(data))
        em.saveToCassandra("ap", "trip_summary_data",SomeColumns( "service_id" ,"asset_id","summ_typ","summ_dt","trp_summ_id","asset_serial_no","avg_sp","c_dist","c_epa","c_gal","c_mil","device_id","device_serial_no","dist","en_addr","en_dt","en_lat","en_long","epa","gal","h_dist","h_epa","h_gal","h_mil","id_tm","max_sp","mil","rec_crt_dt","st_addr","st_lat","st_long","tr_dis","tr_dt","tr_dur"))
  }

def dashBoardInsert(data: TripHistoryData) {
   var em=sc.parallelize(Seq(data)) 
    em.saveToCassandra("ap", "asset_dashboard_data",SomeColumns("service_id","asset_id","hlth_typ","hlth_s_typ","asset_serial_no" as "assetSerialNo","dsh_nval_01","dsh_nval_02","dsh_val_01","dsh_val_02","hlth_col_ind","lst_rfr_dt","rec_crt_dt" ) )    
        }

def dashBoardUpdate(data: TripHistoryData) {

         var em=sc.parallelize(Seq(data))
         em.saveToCassandra("ap", "asset_dashboard_data",SomeColumns("service_id","asset_id","hlth_typ","hlth_s_typ","asset_serial_no" as "assetSerialNo","dsh_nval_01","dsh_nval_02","dsh_val_01","dsh_val_02","hlth_col_ind","lst_rfr_dt","rec_crt_dt" ) )    
        }

错误堆栈跟踪：

18/09/26 21:27:41 ERROR app.ProcessMPacket$: error for processing this event For M-packet
java.io.IOException: Couldn't find ap.asset_dashboard_data or any similarly named keyspace and table pairs
    at com.datastax.spark.connector.cql.Schema$.tableFromCassandra(Schema.scala:358)
    at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:379)
    at com.datastax.spark.connector.RDDFunctions.saveToCassandra(RDDFunctions.scala:35)
    at com.vzt.afm.hum.dh.util.CassandraUtils$.dashBoardInsert(CassandraUtils.scala:275)
    at com.vzt.afm.hum.dh.app.TripAggregation$.updateOdometer(TripAggregation.scala:86)
    at com.vzt.afm.hum.dh.app.ProcessMPacket$$anonfun$1$$anonfun$apply$1.apply(ProcessMPacket.scala:176)
    at com.vzt.afm.hum.dh.app.ProcessMPacket$$anonfun$1$$anonfun$apply$1.apply(ProcessMPacket.scala:129)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
    at scala.collection.mutable.ListBuffer.foreach(ListBuffer.scala:45)
    at com.vzt.afm.hum.dh.app.ProcessMPacket$$anonfun$1.apply(ProcessMPacket.scala:129)
    at com.vzt.afm.hum.dh.app.ProcessMPacket$$anonfun$1.apply(ProcessMPacket.scala:75)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:800)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:800)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

spark cassandra连接器的文档中对此进行了描述。基本上，您需要创建单独的示例 CassandraConnector 类，具有不同的cassandra相关配置，至少不同 spark.cassandra.connection.host ，然后重新定义 c 配置正确。以下是文档中的示例：

val connectorToClusterOne = CassandraConnector(sc.getConf.set(
     "spark.cassandra.connection.host", "127.0.0.1"))
val connectorToClusterTwo = CassandraConnector(sc.getConf.set(
     "spark.cassandra.connection.host", "127.0.0.2"))

val rddFromClusterOne = {
  // Sets connectorToClusterOne as default connection for everything in this code block
  implicit val c = connectorToClusterOne
  sc.cassandraTable("ks","tab")
}

{
  //Sets connectorToClusterTwo as the default connection for everything in this code block
  implicit val c = connectorToClusterTwo
  rddFromClusterOne.saveToCassandra("ks","tab")
}

如果您使用的是dataframes，那么就更容易了，因为您可以在操作级别上指定cluster（示例摘自russelspitzer的博客文章）：

sqlContext.setConf("ClusterOne/spark.cassandra.connection.host", "127.0.0.1")
sqlContext.setConf("ClusterTwo/spark.cassandra.connection.host", "127.0.0.2")

//Read from ClusterOne
val dfFromClusterOne = sqlContext.read.format("org.apache.spark.sql.cassandra")
  .options(Map("cluster" -> "ClusterOne", "keyspace" -> "ks", "table" -> "tab"))
  .load

//Write to ClusterTwo
dfFromClusterOne.write.format("org.apache.spark.sql.cassandra")
  .options(Map("cluster" -> "ClusterTwo", "keyspace" -> "ks", "table" -> "tab"))
  .save

如何将spark与多个cassandra主机的访问和不同的表进行集成，以处理不同的方法

1条答案

相关问题

热门标签

最新问答