我试图在Cassandra的一小部分数据上运行一个火花工作 . 我手头有一个密钥RDD（分区和集群列），我想只在那些密钥上运行我的工作 .
type CassandraKey = (String, String, String, String)
val columns = SomeColumns(ColumnName("pkey1"),ColumnName("pkey2"),ColumnName("pkey3"),ColumnName("ckey1"))
val repartitionedKeys: CassandraPartitionedRDD[CassandraKey] = keys.repartitionByCassandraReplica("keyspace", "table", partitionKeyMapper = columns)
val selectedRows: CassandraJoinRDD[CassandraKey, CassandraRow] =
repartitionedKeys.joinWithCassandraTable[CassandraRow](keyspace, table).on(joinColumns = columns)
selectedRows.collect()
我在BoundStatementBuilder收到以下错误：19
java.lang.IllegalArgumentException: ckey1 is not a column defined in this metadata
我的表架构如下：
CREATE TABLE "keyspace".table (
pkey1 text,
pkey2 text,
pkey3 text,
ckey1 text,
ckey2 text,
ckey3 timestamp,
data text,
PRIMARY KEY (( pkey1, pkey2, pkey3 ), ckey1, ckey2, ckey3)
)
查看代码我可以看到在BoundStatementBuilder中，columnTypes正在从ReplicaLocator.keyByReplicas启动的虚拟查询中解析 . 此查询用于从表中检索分区令牌，并且仅在分区键上构造where子句 .
另外我可以在RDDFunction.repartitionByCassandraReplica：183中看到给定的partitionKeyMapper被忽略，但这似乎不会引起任何问题 .
我正在使用连接器版本1.5.1
1 回答
"Repartition"部分只能在分区键上，因此不要在那里指定列，或者只选择分区键列 . 仅使用
joinWithCassandraTable调用指定所有连接列 .