我正在使用cassandra 2.0.3,我想使用pyspark(Apache Spark Python API)从cassandra数据创建一个RDD对象.
请注意:我不想从pyspark API执行导入CQL和CQL查询,而是想创建一个RDD,我想在其上做一些转换.
我知道这可以在Scala中完成,但我无法找到如何通过pyspark完成.
真的很感激,如果有人能指导我这个.
这是关于我从Apache Spark查询Cassandra时遇到的问题.
来自Spark的正常查询工作正常,没有任何问题,但是当我查询条件是关键时,我得到以下错误.最初我尝试查询复合键列族,它也给出了与下面相同的问题.
"引起:InvalidRequestException(为什么:如果包含Equal,则empid不能被多个关系限制)"
专栏系列:
CREATE TABLE emp (
empID int,
deptID int,
first_name varchar,
last_name varchar,
PRIMARY KEY (empID));
列族内容:
empID, deptID, first_name, last_name
104, 15, 'jane', 'smith'
示例SCALA代码:
val job=new Job()
job.setInputFormatClass(classOf[CqlPagingInputFormat])
val host: String = "localhost"
val port: String = "9160"
ConfigHelper.setInputInitialAddress(job.getConfiguration(), host)
ConfigHelper.setInputRpcPort(job.getConfiguration(), port)
ConfigHelper.setInputColumnFamily(job.getConfiguration(), "demodb", "emp")
ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner")
CqlConfigHelper.setInputColumns(job.getConfiguration(), "empid,deptid,first_name,last_name")
//CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), limit.toString)
CqlConfigHelper.setInputWhereClauses(job.getConfiguration(),"empid='104'")
// Make a new Hadoop RDD
val casRdd = sc.newAPIHadoopRDD(job.getConfiguration(),
classOf[CqlPagingInputFormat],
classOf[Map[String, ByteBuffer]],
classOf[Map[String, ByteBuffer]])
我恳请你告诉我,如果有任何解决这种情况,因为我在过去几天努力克服这个问题.
谢谢
我正在运行cassandra 1.2.6和Apache Spark 0.8.0.
在这里,我使用Spark的newAPIHadoopRDD从Cassandra创建一个RDD.
当我从spark appln运行作业时,我发现cassandra正在记录以下错误消息
INFO 21:36:28,629 Listening for thrift clients...
DEBUG 21:36:29,730 Disseminating load info ...
DEBUG 21:36:57,844 Started replayAllFailedBatches
DEBUG 21:36:57,845 forceFlush requested but everything is clean in batchlog
DEBUG 21:36:57,846 Finished replayAllFailedBatches
DEBUG 21:37:29,731 Disseminating load info ...
DEBUG 21:37:57,846 Started replayAllFailedBatches
DEBUG 21:37:57,847 forceFlush requested but everything is clean in batchlog
DEBUG 21:37:57,847 Finished replayAllFailedBatches
DEBUG 21:38:29,732 Disseminating load info ...
DEBUG 21:38:57,847 Started replayAllFailedBatches
DEBUG 21:38:57,849 forceFlush requested but everything is clean in …