这是关于我在从apachespark查询cassandra时遇到的一个问题。
spark的正常查询工作正常,没有任何问题,但是当我使用关键条件进行查询时,会出现以下错误。最初,我尝试查询复合键列族,它也给出了与下面相同的问题。
“原因:invalidrequestexception(why:empid cannot 如果包含一个相等的关系,则受多个关系的限制
柱族:
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]])
我恳请您让我知道,如果有任何工作,为这种情况下,因为我正在努力克服这个问题在过去几天。
谢谢
1条答案
按热度按时间cgh8pdjw1#
出现此错误的原因是查询在cassandra中的翻译方式(检查
org.apache.cassandra.hadoop.cql3.CqlPagingRecordReader#whereClause
详细信息)。将查询转换为cassandra时,它具有以下语法:从“emp”中选择*where token(“empid”)>?和令牌(“empid”)<=?empid='104'限制1000允许过滤
有一个相关的jira(cassandra-6151)标记as不会修复,它讨论了一些解决问题的方法。从我找到的小文件来看,
CqlConfigHelper.setInputWhereClauses
只能用于不属于键的索引列。希望对你有帮助。