title: Joining Kafka and Cassandra DataFrames in Spark Streaming ignores C* predicate pushdown
body: <p><strong>Intent</strong></p>
  <p>I'm receiving data from Kafka via direct stream and would like to enrich the messages with data from Cassandra. The Kafka messages (Protobufs) are decoded into DataFrames and then joined with a (supposedly pre-filtered) DF from Cassandra. The relation of (Kafka) streaming batch size to raw C* data is [several streaming messages to millions of C* rows], BUT the join always yields exactly ONE result [1:1] per message. After the join the resulting DF is eventually stored to another C* table.</p>
  <p><strong>Problem</strong></p>
  <p>Even though I'm joining the two DFs on the full Cassandra primary key and pushing the corresponding filter to C*, it seems that Spark is loading the whole C* data-set into memory before actually joining (which I'd like to prevent by using the filter/predicate pushdown). This leads to a lot of shuffling and tasks being spawned, hence the &quot;simple&quot; join takes forever...</p>
  <pre class="lang-scala prettyprint-override"><code>def main(args: Array[String]) {
      val conf = new SparkConf()
        .setAppName(&quot;test&quot;)      
        .set(&quot;spark.cassandra.connection.host&quot;, &quot;xxx&quot;)
        .set(&quot;spark.cassandra.connection.keep_alive_ms&quot;, &quot;30000&quot;)
        .setMaster(&quot;local[*]&quot;)
        
      val ssc = new StreamingContext(conf, Seconds(10))
      ssc.sparkContext.setLogLevel(&quot;INFO&quot;)
      
      // Initialise Kafka
      val kafkaTopics = Set[String](&quot;xxx&quot;)
      val kafkaParams = Map[String, String](
        &quot;metadata.broker.list&quot; -&gt; &quot;xxx:32000,xxx:32000,xxx:32000,xxx:32000&quot;,
        &quot;auto.offset.reset&quot; -&gt; &quot;smallest&quot;)
      
      // Kafka stream
      val messages = KafkaUtils.createDirectStream[String, MyMsg, StringDecoder, MyMsgDecoder](ssc, kafkaParams, kafkaTopics)      
      
      // Executed on the driver
      messages.foreachRDD { rdd =&gt;
        
        // Create an instance of SQLContext
        val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
        import sqlContext.implicits._
        
        // Map MyMsg RDD
        val MyMsgRdd = rdd.map{case (key, MyMsg) =&gt; (MyMsg)}
        
        // Convert RDD[MyMsg] to DataFrame
        val MyMsgDf = MyMsgRdd.toDF()        
          .select(
              $&quot;prim1Id&quot; as 'prim1_id,
              $&quot;prim2Id&quot; as 'prim2_id,
              $...
        )
        
        // Load DataFrame from C* data-source
        val base_data = base_data_df.getInstance(sqlContext)    
        
        // Left join on prim1Id and prim2Id
        val joinedDf = MyMsgDf.join(base_data,
              MyMsgDf(&quot;prim1_id&quot;) === base_data(&quot;prim1_id&quot;) &amp;&amp;
              MyMsgDf(&quot;prim2_id&quot;) === base_data(&quot;prim2_id&quot;), &quot;left&quot;)
              .filter(base_data(&quot;prim1_id&quot;).isin(MyMsgDf(&quot;prim1_id&quot;))
                  &amp;&amp; base_data(&quot;prim2_id&quot;).isin(MyMsgDf(&quot;prim2_id&quot;)))          
                  
        joinedDf.show()
        joinedDf.printSchema()
        
        // Select relevant fields
              
        // Persist
      }
      
      // Start the computation
      ssc.start()
      ssc.awaitTermination()
  }
  </code></pre>
  <p><strong>Environment</strong></p>
  <ul>
  <li>Spark 1.6</li>
  <li>Cassandra 2.1.12</li>
  <li>Cassandra-Spark-Connector 1.5-RC1</li>
  <li>Kafka 0.8.2.2</li>
  </ul>
  <p><strong>SOLUTION</strong></p>
  <p>From discussions on the DataStax Spark Connector for Apache Cassandra ML</p>
  <ul>
  <li><a href="https://groups.google.com/a/lists.datastax.com/forum/#!topic/spark-connector-user/Itt8lpkaGhw" rel="nofollow noreferrer">Joining Kafka and Cassandra DataFrames in Spark Streaming ignores C* predicate pushdown</a></li>
  <li><a href="https://groups.google.com/a/lists.datastax.com/forum/#!topic/spark-connector-user/CZZCQncgCQM" rel="nofollow noreferrer">How to create a DF from CassandraJoinRDD</a></li>
  </ul>
  <p>I've learned the following:</p>
  <blockquote>
  <p>Quoting Russell Spitzer</p>
  <ol>
  <li><p>This wouldn't be a case of predicate pushdown. This is a join on a partition key column. Currently only joinWithCassandraTable supports this direct kind of join although we are working on some methods to try to have this automatically done within Spark.</p>
  </li>
  <li><p>Dataframes can be created from any RDD which can have a schema applied to it. The easiest thing to do is probably to map your joinedRDD[x,y] to Rdd[JoinedCaseClass] and then call toDF (which will require importing your sqlContext implicits.) See the <a href="http://spark.apache.org/docs/latest/sql-programming-guide.html#inferring-the-schema-using-reflection" rel="nofollow noreferrer">DataFrames documentation </a> here for more info.</p>
  </li>
  </ol>
  </blockquote>
  <p>So the actual implementation now resembles something like</p>
  <pre class="lang-scala prettyprint-override"><code>// Join myMsg RDD with myCassandraTable
  val joinedMsgRdd = myMsgRdd.joinWithCassandraTable(
    &quot;keyspace&quot;,
    &quot;myCassandraTable&quot;,
    AllColumns,
    SomeColumns(
        &quot;prim1_id&quot;,
        &quot;prim2_id&quot;
    )
  ).map{case (myMsg, cassandraRow) =&gt; 
  
    JoinedMsg(
      foo = myMsg.foo
      bar = cassandraRow.bar
    )
  }
  
  // Convert RDD[JoinedMsg] to DataFrame
  val myJoinedDf = joinedMsgRdd.toDF()
  </code></pre>