分区键部分 url 的空值无效

Question

我有以下代码试图在 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(Responses.java:103) 在 com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResultSetFuture.java:140) 在 com.datastax.driver.core.RequestHandler.setFinalResult(RequestHandler.java:293) 在 com.datastax.driver.core.RequestHandler.onSet(RequestHandler.java:455) 在 com.datastax.driver.core.Connection$Dispatcher.messageReceived(Connection.java:734) 在 org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) 在 org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler.handleUpstream(IdleStateAwareChannelUpstreamHandler.java:36) 在 org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream（DefaultChannelPipeline.java:564） 在 org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) 在 org.jboss.netty.handler.timeout.IdleStateHandler.messageReceived(IdleStateHandler.java:294) 在 org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) 在 org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream（DefaultChannelPipeline.java:564） 在 org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.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（DefaultChannelPipeline.java:564） 在 org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) 在 org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) 在 org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462) 在 org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443) 在 org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303) 在 org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) 在 org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream（DefaultChannelPipeline.java:564） 在 org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream（DefaultChannelPipeline.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(ThreadRenamingRunnable.java:108) 在 org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) ... 3 更多

Answer 1

简单，异常告诉你它无法执行连接，因为用于连接 ImageMetadata 和 PagesMetadata 的列为空。

在您的情况下，ImageMetadata 中的一些 url (pu) 值为 null。

奇怪的是你用 url 可以为空 (Option[String]) 定义 PagesMetadata 并且它似乎是表的主键的一部分

使其发挥作用的一种解决方案是：

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