title: Read CSV File in Spark and Write it to Cassandra
body: <p>I am trying to read a CVS File with Spark and then save it to Cassandra. Saving to Cassandra is working, when I'm using trivial values.</p>
  
  <p>I have a file with the following values:</p>
  
  <p><code>id,name,tag1|tag2|tag3</code></p>
  
  <p>I want to store it in a cassandra table: </p>
  
  <p><code>id bigint, name varchar, tags set</code></p>
  
  <p>I defined a case class for this:</p>
  
  <p><code>case class Item(id:Integer,name:String,tag:Set[String])</code></p>
  
  <p>Then I use this expression for getting the RDD out of the CVS file</p>
  
  <p><code>val items = sc.textFile("items.csv").map(l =&gt; l.split(",") match {case Array (a,b,c) =&gt; Item(Integer.parseInt(a),b,c.split("\\|").toSet)})</code></p>
  
  <p>When I now call <code>collect</code> or <code>saveToCassandra</code> on items (which starts the processing) I get the following error:</p>
  
  <p><code>org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 29.0 failed 1 times, most recent failure: Lost task 1.0 in stage 29.0 (TID 38, localhost): scala.MatchError: [Ljava.lang.String;@6030bbe6 (of class [Ljava.lang.String;)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(&lt;console&gt;:33)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(&lt;console&gt;:33)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
      at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:249)
      at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:172)
      at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:79)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:242)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
      at org.apache.spark.scheduler.Task.run(Task.scala:64)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  </code></p>