查询 Cassandra 表以获取每个 Kafka 消息

Question

我正在尝试查询卡桑德拉表以获取每条卡夫卡消息。

下面是我一直在处理的代码：

 def main(args: Array[String]) {
 val spark = SparkSession
  .builder()
  .master("local[*]")
  .appName("Spark SQL basic example")
  .config("spark.cassandra.connection.host", "localhost")
  .config("spark.cassandra.connection.port", "9042")
  .getOrCreate()
val topicsSet = List("Test").toSet
val kafkaParams = Map[String, Object](
          "bootstrap.servers" -> "localhost:9092",
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> "12345",
          "auto.offset.reset" -> "latest",
          "enable.auto.commit" -> (false: java.lang.Boolean)
          )
val messages = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))
val lines = messages.map(_.value)
val lines_myobjects = lines.map(line =>
  new Gson().fromJson(line, classOf[myClass]) // The myClass is a simple case class which extends serializable
//This changes every single message into an object
)

现在事情变得复杂了，我无法绕过可以查询与 kafka 消息中的消息相关的 cassandra 表的点。每个 kafka 消息对象都有一个返回方法。

我已经尝试了多种方法来解决这个问题。例如：

val transformed_data = lines_myobjects.map(myobject => {
   val forest = spark.read
    .format("org.apache.spark.sql.cassandra")
    .options(Map( "table" -> "mytable", "keyspace" -> "mydb"))
    .load()
    .filter("userid='" + myobject.getuserId + "'")
)}

我也尝试过ssc.cassandraTable没有给我运气。

主要目标是从数据库中获取用户标识与来自 kafka 消息的用户标识匹配的所有行。

我想提到的一件事是，即使每次加载或查询 cassandra 数据库效率不高，但 cassandra 数据库每次都会更改。

Answer 1

你不能在.map(里面做spark.read或ssc.cassandraTable。因为这意味着您将尝试为每条消息创建新的RDD。它不应该那样工作。

请选择以下选项：

1 - 如果可以通过一个/两个 CQL 查询询问所需的数据，请尝试在.mapPartitions(中使用 CassandraConnector。像这样：

import com.datastax.spark.connector._
import com.datastax.spark.connector.cql._
val connector = ...instantiate CassandraConnector onece here
val transformed_data = lines_myobjects.mapPartitions(it => {
   connector.withSessionDo { session =>
       it.map(myobject => session.execute("CQL QUERY TO GET YOUR DATA HERE", myobject.getuserId)
})

2 - 否则(如果按主键/分区键选择(请考虑.joinWithCassandraTable。像这样：

import com.datastax.spark.connector._
val mytableRDD = sc.cassandraTable("mydb", "mytable")
val transformed_data = lines_myobjects
    .map(myobject => {
       Tuple1(myobject.getuserId) // you need to wrap ids to a tuple to do join with Cassandra
    })
    .joinWithCassandraTable("mydb", "mytable")
    // process results here

Answer 2

我会以不同的方式解决这个问题。流入 Cassandra 的数据通过 Kafka 路由(并从 Kafka 发送到带有 Kafka Connect 接收器的 Cassandra(。使用Kafka中的数据，您可以在数据流之间加入，无论是在Spark中，还是使用Kafka的Streams API或KSQL。Kafka Streams 和 KSQL 都支持您在此处执行的流表连接。您可以在此处和此处看到它与 KSQL 的实际操作。