如何将Cassandra ResultSet转换为Spark DataFrame？

Question

我通常会从Cassandra的数据加载到Apache的星火使用Java这样：

SparkContext sparkContext = StorakleSparkConfig.getSparkContext(); 

CassandraSQLContext sqlContext = new CassandraSQLContext(sparkContext); 
    sqlContext.setKeyspace("midatabase"); 

DataFrame customers = sqlContext.cassandraSql("SELECT email, first_name, last_name FROM store_customer " + 
      "WHERE CAST(store_id as string) = '" + storeId + "'"); 

但是想象一下，我有一个sharder，我需要几个partion密钥加载到该数据帧。我可以在我的查询中使用WHERE IN（...）并再次使用cassandraSql方法。但是由于在协调器节点方面存在单点故障的臭名昭着的问题，我有点不情愿使用WHERE IN。这是在这里解释说：

https://lostechies.com/ryansvihla/2014/09/22/cassandra-query-patterns-not-using-the-in-query-for-multiple-partitions/

是否有使用几个查询，但它们加载到一个单一的数据帧的方法吗？

Answer 1

这样做的一种方法是运行单个查询和unionAll/union多个DataFrames/RDDs。

SparkContext sparkContext = StorakleSparkConfig.getSparkContext(); 

CassandraSQLContext sqlContext = new CassandraSQLContext(sparkContext); 
    sqlContext.setKeyspace("midatabase"); 

DataFrame customersOne = sqlContext.cassandraSql("SELECT email, first_name, last_name FROM store_customer " + "WHERE CAST(store_id as string) = '" + storeId1 + "'"); 

DataFrame customersTwo = sqlContext.cassandraSql("SELECT email, first_name, last_name FROM store_customer " + "WHERE CAST(store_id as string) = '" + storeId2 + "'"); 

DataFrame allCustomers = customersOne.unionAll(CustomersTwo)