基于cassandra表分区密钥的spark数据存储

Question

null

• 火花:1.3
• 卡桑德拉:2.1
• 连接器:1.3.1
• 火花节点（5）和CASS*集群节点（4）运行在不同的数据中心

代码提取。请使用以上链接下载代码以获取更多详细信息

步骤1：将数据加载到8个spark分区中

Map<String, String> map = new HashMap<String, String>();
CassandraTableScanJavaRDD<TestTable> tableRdd = javaFunctions(conf)
 .cassandraTable("testkeyspace", "testtable", mapRowTo(TestTable.class, map));

.repartitionByCassandraReplica(
        "testkeyspace",
        "testtable",
        partitionNumPerHost,
        someColumns("id"),
        mapToRow(TestTable.class, map));

rdd.mapPartitionsWithIndex(...{
@Override
 public Iterator<String> call(..) throws Exception {
 List<String> list = new ArrayList<String>();
 list.add("PartitionId-" + integer);

 while (itr.hasNext()) {
    TestTable value = itr.next();
    list.add(Integer.toString(value.getId()));
 }
 return list.iterator();
}
}, true).collect();

加载Rdd值

----------------------------
Table load - PartitionId -1
----------------------------
15
22

--------------------------------------
Repartitioned values - PartitionId -1
--------------------------------------
33
16

Answer 1

由Cassandra replica进行的重新分区并不确定地放置键。目前有一张票可以改变这一点。

https://datastax-oss.atlassian.net/projects/sparkc/issues/sparkc-278

现在的一个解决办法是将Partitionspernode参数设置为1。