代码之家 › 专栏 › 技术社区 › l0n3r4n83r

分区密钥部分url的空值无效

spark-cassandra-connector cql3 cassandra-2.0 cassandra apache-spark

l0n3r4n83r · 技术社区 · 9 年前

我有下面的代码,它试图在spark中加入2个cassandra表。

 val imageKeywords = sc.cassandraTable[ImageMetadata]("images", "metadata")
 val imageAndPageKeywords = imageKeywords
  .joinWithCassandraTable[PagesMetadata]("pages2", "metadata")
  .on(SomeColumns("tid", "url" as "pu"))

我用来映射数据的案例类如下

case class ImageMetadata(tid: String, iu: String, pu: Option[String],
mk: List[String], fk: List[String], ak: List[String], ipk: List[String], pk: List[String], ik: List[String], ck: List[String])

case class PagesMetadata(tid: String, url: String, pk: List[String], uk: List[String], hk: List[String], ok: List[String], tc: List[String])

当我尝试执行以下操作时出错

imageAndPageKeywords.collect.toList.sortBy(_._1.tid).take(10).foreach(println)

错误堆栈跟踪-

由:com.datastax.driver.core.exceptions引起。InvalidQueryException:分区密钥部分url的空值无效位于com.datastax.driver.core.Responses$Error.asException(Response.java:103) 位于com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResaultSetFuture.java:140) 位于com.datastax.driver.core.RequestHandler.setFinalResult(请求处理程序.java:293) 位于com.datastax.driver.core.RequestHandler.onSet(RequestHander.java:455) 位于com.datastax.driver.core.Connection$Dispatcher.messageReceived(Connection.java:734) 在org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannel上游处理程序.java:70) 位于org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler.handleUpstream(IdleStateAware通道上游处理程序.java:36) org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannel管道.java:564) org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannel HandlerContext.sendUpstream(DefaultChanne管道.java:791) org.jboss.netty.handler.timeout.IdleStateHandler.messageReceived(IdleStateHandler.java:294) 在org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannel上游处理程序.java:70) org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannel管道.java:564) org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannel HandlerContext.sendUpstream(DefaultChanne管道.java:791) org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) 在org.jboss.netty.handler.codec.oneone.OneToOneDecoder.handleUpstream(OneToOneDecoder.java:70) org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannel管道.java:564) org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannel HandlerContext.sendUpstream(DefaultChanne管道.java:791) org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) 在org.jboss.netty.handler.codec.frame.FrameDecoder.unloadAndFireMessageReceived(FrameDeoder.java:462) org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDeoder.java:443) 在org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDeoder.java:303) 在org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannel上游处理程序.java:70) org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannel管道.java:564) org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannel管道.java:559) org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) 在org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) 在org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108) org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:318) org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89) 在org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamengRunnable.java:108) 在org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLock ProofWorker.java:42) …还有3个

1 回复 | 直到 9 年前

doanduyhai 9 年前

简单地说,异常告诉您它无法执行联接,因为用于联接的列 图像元数据 具有 页面元数据 为空。

在你的情况下,一些网址 (pu)值 图像元数据 为空。

奇怪的是,你定义了 页面元数据 具有网址可为空(Option[String]),并且它似乎是表主键的一部分

实现这一目标的一个解决方案是:

val imageAndPageKeywords = imageKeywords
  .filter(im -> im.pu.isDefined)
  .joinWithCassandraTable[PagesMetadata]("pages2", "metadata")
  .on(SomeColumns("tid", "url" as "pu"))