如何將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)