Scala：通过 DStream 拆分来自 kafka 的数据答案

【问题标题】：Scala: Splitting the data coming from kafka vi a DStreamScala：通过 DStream 拆分来自 kafka 的数据
【发布时间】：2018-12-15 00:11:24
【问题描述】：

我正在以

的形式从 kafka 接收数据

{"email":"test@example","firstname":"Example","lastname":"User"}

我想访问电子邮件 ID 和名字，并希望将其与来自 cassandra 的数据进行比较：

CassandraRow{email: abc@xyz.com}

【问题讨论】：

您能否扩展您的问题 - 这个比较的结果应该是什么？你如何从 Cassandra 获取数据？ Cassandra 表的架构是什么？
我想比较凭据（电子邮件 ID 和姓名），如果它们相同，我想向 kafka 主题发送一条消息，表明它们是相等的。我正在使用 SparkStreaming 的变量从 cassandra 表中获取数据，其中行 val data1 = ssc.cassandraTable("test","login").select("email","name","lastname") .where("email =?","abc@xyz.com") val rddQueue = new Queue[RDD[com.datastax.spark.connector.CassandraRow]]() val dstream = ssc.queueStream(rddQueue) 我的 cassandra 表也有电子邮件条目身份证、姓名和姓氏
表的主键是什么？
电子邮件是主键
请将您在评论中发布的额外信息添加到实际问题中

【解决方案1】：

您需要使用 joinWithCassandraTable function 与 Cassandra 执行联接...

为了更有效，您可能需要对从 Kafka 获得的 RDD 重新分区，以匹配 Cassandra 表中的分区。代码可能如下所示：

val resultRdd = kafkaRDD.repartitionByCassandraReplica("ks","emails")
   .joinWithCassandraTable("ks","emails")

然后您可以分析，如果名称匹配等。加入后，您应该只获得 Cassandra 中有电子邮件的记录...

【讨论】：

val lines=KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics).map(_._2)。我从kafka得到这个directStream所以我应该如何重新分区这个RDD以应用joinwithcassandraTable操作以及变量resultRdd的值是什么形式