title: Keeping data together in spark based on cassandra table partition key
body: <p>When loading data from Cassandra table, a spark partition represents all rows with same partition key. However, when I create data in spark with same partition key and re-partitioning the new RDD  using .repartitionByCassandraReplica(..) method, it ends up in a different spark partition? How do I achieve consistent partitions in spark using the partition-scheme defined by the Spark-Cassandra connector?</p>
  
  <blockquote>
    <p>Links to download CQL and Spark job code that I tested</p>
  </blockquote>
  
  <ul>
  <li><a href="https://sarmathi.blob.core.windows.net/spark/test.cql" rel="nofollow">.CQL with the keyspace and table schema</a>. </li>
  <li><a href="https://sarmathi.blob.core.windows.net/spark/CassandraConnectorTest.java" rel="nofollow">Spark job</a> and other classes.</li>
  </ul>
  
  <blockquote>
    <p>Version and other information</p>
  </blockquote>
  
  <ul>
  <li>Spark : 1.3</li>
  <li>Cassandra : 2.1</li>
  <li>Connector : 1.3.1</li>
  <li>Spark nodes (5) and Cass* cluster nodes (4) runs in different data centers</li>
  </ul>
  
  <blockquote>
    <p>Code extract. Download code using above links for more details</p>
  </blockquote>
  
  <p><strong>Step 1 : Loads data into 8 spark partitions</strong></p>
  
  <pre><code>Map&lt;String, String&gt; map = new HashMap&lt;String, String&gt;();
  CassandraTableScanJavaRDD&lt;TestTable&gt; tableRdd = javaFunctions(conf)
   .cassandraTable("testkeyspace", "testtable", mapRowTo(TestTable.class, map));
  </code></pre>
  
  <p><strong>Step 2 : Repartition data into 8 partitions</strong></p>
  
  <pre><code>.repartitionByCassandraReplica(
          "testkeyspace",
          "testtable",
          partitionNumPerHost,
          someColumns("id"),
          mapToRow(TestTable.class, map));
  </code></pre>
  
  <p><strong>Step 3: Print partition id and values for both rdds</strong></p>
  
  <pre><code>rdd.mapPartitionsWithIndex(...{
  @Override
   public Iterator&lt;String&gt; call(..) throws Exception {
   List&lt;String&gt; list = new ArrayList&lt;String&gt;();
   list.add("PartitionId-" + integer);
  
   while (itr.hasNext()) {
      TestTable value = itr.next();
      list.add(Integer.toString(value.getId()));
   }
   return list.iterator();
  }
  }, true).collect();
  </code></pre>
  
  <p><strong>Step 4 : Snapshot of results printed on Partition 1. Different for both Rdds but expect to be same</strong></p>
  
  <p>Load Rdd values</p>
  
  <pre><code>----------------------------
  Table load - PartitionId -1
  ----------------------------
  15
  22
  
  --------------------------------------
  Repartitioned values - PartitionId -1
  --------------------------------------
  33
  16
  </code></pre>