title: Writing data to cassandra using spark
body: <p>I have a spark job written in Scala, in which I am just trying to write one line separated by commas, coming from Kafka producer to Cassandra database. <strong>But I couldn't call saveToCassandra.</strong> 
  I saw few examples of wordcount where they are writing map structure to Cassandra table with two columns and it seems working fine. But I have many columns and I found that the data structure needs to parallelized. 
  Here's is the sample of my code:</p>
  
  <pre><code>object TestPushToCassandra extends SparkStreamingJob {
  def validate(ssc: StreamingContext, config: Config): SparkJobValidation = SparkJobValid
  
  def runJob(ssc: StreamingContext, config: Config): Any = {
  
  val bp_conf=BpHooksUtils.getSparkConf()
  val brokers=bp_conf.get("bp_kafka_brokers","unknown_default")
  
  
  val input_topics = config.getString("topics.in").split(",").toSet
  
  
  val output_topic = config.getString("topic.out")
  
  
  val kafkaParams = Map[String, String]("metadata.broker.list" -&gt; brokers)
  val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, input_topics)
  
  
  val lines = messages.map(_._2)
  val words = lines.flatMap(_.split(","))
  
  val li = words.par
  
  li.saveToCassandra("testspark","table1", SomeColumns("col1","col2","col3"))
  li.print()
  
  
  
  words.foreachRDD(rdd =&gt;
    rdd.foreachPartition(partition =&gt;
      partition.foreach{
        case x:String=&gt;{
  
          val props = new HashMap[String, Object]()
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer")
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer")
  
          val outMsg=x+" from spark"
          val producer = new KafkaProducer[String,String](props)
          val message=new ProducerRecord[String, String](output_topic,null,outMsg)
          producer.send(message)
        }
      }
  
  
    )
  )
  
  
  ssc.start()
  ssc.awaitTermination()
  }
  }
  </code></pre>
  
  <p>I think it's the syntax of Scala that I am not getting correct.
  Thanks in advance.</p>