title: Spark Cassandra Connector: for comprehension error (type mismatch)
body: <h2>Problem</h2>
  
  <p>Maybe this is due to my lack of Scala knowledge, but it seems like adding another level to the <em>for comprehension</em> should just work. If the first <em>for comprehension</em> line is commented out, the code works. I ultimately want a Set[Int] instead of '1 to 2', but it serves to show the problem. The first two lines of the <em>for</em> should not need a type specifier, but I include it to show that I've tried the obvious.</p>
  
  <h2>Tools/Jars</h2>
  
  <ul>
  <li>IntelliJ 2016.1</li>
  <li>Java 8</li>
  <li>Scala 2.10.5</li>
  <li>Cassandra 3.x</li>
  <li>spark-assembly-1.6.0-hadoop2.6.0.jar (pre-built)</li>
  <li>spark-cassandra-connector_2.10-1.6.0-M1-SNAPSHOT.jar (pre-built)</li>
  <li>spark-cassandra-connector-assembly-1.6.0-M1-SNAPSHOT.jar (I built)</li>
  </ul>
  
  <h2>Code</h2>
  
  <pre><code>case class NotifHist(intnotifhistid:Int, eventhistids:Seq[Int], yosemiteid:String, initiatorname:String)
  case class NotifHistSingle(intnotifhistid:Int, inteventhistid:Int, dataCenter:String, initiatorname:String)
  
  object SparkCassandraConnectorJoins {
    def joinQueryAfterMakingExpandedRdd(sc:SparkContext, orgNodeId:Int) {
  
    val notifHist:RDD[NotifHistSingle] = for {
      orgNodeId:Int &lt;- 1 to 2   // comment out this line and it works
      notifHist:NotifHist &lt;- sc.cassandraTable[NotifHist](keyspace, "notifhist").where("intorgnodeid = ?", orgNodeId)
      eventHistId &lt;- notifHist.eventhistids
    } yield NotifHistSingle(notifHist.intnotifhistid, eventHistId, notifHist.yosemiteid, notifHist.initiatorname)
    ...etc...
   }
  </code></pre>
  
  <h2>Compilation Output</h2>
  
  <pre><code>Information:3/29/16 8:52 AM - Compilation completed with 1 error and 0 warnings in 1s 507ms
         /home/jpowell/Projects/SparkCassandraConnector/src/com/mir3/spark/SparkCassandraConnectorJoins.scala
  **Error:(88, 21) type mismatch;
   found   : scala.collection.immutable.IndexedSeq[Nothing]
   required: org.apache.spark.rdd.RDD[com.mir3.spark.NotifHistSingle]
        orgNodeId:Int &lt;- 1 to 2
                      ^**
  </code></pre>
  
  <h2>Later</h2>
  
  <p>@slouc Thanks for the comprehensive answer. I was using the <em>for comprehension's</em> syntactic sugar to also keep state from the second statement to fill elements in the NotifHistSingle ctor, so I don't see how to get the equivalent map/flatmap to work. Therefore, I went with the following solution:</p>
  
  <pre><code>def joinQueryAfterMakingExpandedRdd(sc:SparkContext, orgNodeIds:Set[Int]) {
  
    def notifHistForOrg(orgNodeId:Int): RDD[NotifHistSingle] = {
      for {
        notifHist &lt;- sc.cassandraTable[NotifHist](keyspace, "notifhist").where("intorgnodeid = ?", orgNodeId)
        eventHistId &lt;- notifHist.eventhistids
      } yield NotifHistSingle(notifHist.intnotifhistid, eventHistId, notifHist.yosemiteid, notifHist.initiatorname)
    }
    val emptyTable:RDD[NotifHistSingle] = sc.emptyRDD[NotifHistSingle]
    val notifHistForAllOrgs:RDD[NotifHistSingle] = orgNodeIds.foldLeft(emptyTable)((accum, oid) =&gt; accum ++ notifHistForOrg(oid))
  }
  </code></pre>