如何在Spark中以小块的形式迭代大型Cassandra表

Question

在我的测试环境中,我有1个Cassandra节点和3个Spark节点.我想迭代一个大约有200k行的大表,每个大约需要20-50KB.

CREATE TABLE foo (
  uid timeuuid,
  events blob,
  PRIMARY KEY ((uid))
) 

这是在spark集群中执行的scala代码

val rdd = sc.cassandraTable("test", "foo")

// This pulls records in memory, taking ~6.3GB
var count = rdd.select("events").count()  

// Fails nearly immediately with 
// NoHostAvailableException: All host(s) tried for query failed [...]
var events = rdd.select("events").collect()

Cassandra 2.0.9,Spark:1.2.1,Spark-cassandra-connector-1.2.0-alpha2

我试图只运行collect,没有count- 在这种情况下它只是快速失败NoHostAvailableException.

问题:迭代大表读取和一次处理小批量行的正确方法是什么？

Answer 1

Cassandra Spark Connector中有2个设置来调整块大小(将它们放在SparkConf对象中):

• spark.cassandra.input.split.size:每个Spark分区的行数(默认为100000)
• spark.cassandra.input.page.row.size:每个获取页面的行数(即网络往返)(默认为1000)

此外,您不应该collect在示例中使用该操作,因为它将获取驱动程序应用程序内存中的所有行,并可能引发内存不足异常.collect只有在确定它会产生少量行时才能使用该操作.该count行为是不同的,它仅产生整数.所以我建议你像你一样从Cassandra加载你的数据,处理它,然后存储结果(用Cassandra,HDFS,等等).