如何对joinWithCassandraTable生成的CassandraJoinRDD按指定子列表过滤

核心实现思路 #

通过广播变量分发待匹配的用户列表，再对关联后的RDD执行过滤逻辑即可实现需求，针对200万条数据的场景，同步增加缓存、类型转换优化来提升执行效率。 #

具体实现代码 #

步骤1：定义广播变量分发用户列表 #

需要在所有Executor节点的计算任务中用到userlistofitems，用广播变量可以避免全量副本在每个Task中重复传输，大幅降低内存开销：

// 定义待匹配的用户列表
val userlistofitems = Array("jas", "yuk")
// 广播到所有Executor节点
val bcUserList = sc.broadcast(userlistofitems)

步骤2：缓存关联后的RDD（适配多组用户列表场景） #

因为需要对应多组用户列表重复提取数据，建议先将关联结果缓存，避免重复读取文本文件、重复请求Cassandra：

// 原有关联逻辑，最后加上cache持久化
val outrdd = sc.textFile("/usr/local/spark/bin/select_element/src/main/scala/file_small.txt")
  .map(_.toString)
  .map(Tuple1(_))
  .joinWithCassandraTable(settings.keyspace, settings.table)
  .select("id", "listofitems")
  .cache() // 内存足够优先用cache，内存不足可以换成persist(StorageLevel.MEMORY_AND_DISK)

步骤3：执行过滤逻辑 #

将listofitems转成Set后做包含判断，时间复杂度更低，判断逻辑为Cassandra中该行的listofitems需要包含用户列表的所有元素：

val filteredRdd = outrdd.filter { case (_, cassandraRow) =>
  // 从Cassandra行中取出listofitems，转Set提升包含判断效率
  val itemSet = cassandraRow.getList[String]("listofitems").toSet
  // 判断用户列表的所有元素都在itemSet中存在
  bcUserList.value.forall(itemSet.contains)
}

可选优化（过滤条件下压到Cassandra） #

如果你提前为listofitems列创建了Cassandra二级索引或者SAI（存储附加索引），可以将过滤条件下压到Cassandra服务端，减少拉取到Spark侧的数据量：

val outrdd = sc.textFile("/usr/local/spark/bin/select_element/src/main/scala/file_small.txt")
  .map(_.toString)
  .map(Tuple1(_))
  .joinWithCassandraTable(settings.keyspace, settings.table)
  .select("id", "listofitems")
  .where("listofitems contains ? and listofitems contains ?", "jas", "yuk") // 下压过滤条件
  .cache()

内容的提问来源于stack exchange，提问作者Anshita Saxena