Spark UDF使用Cassandra连接器查找键

如何解决Spark UDF使用Cassandra连接器查找键

我正在尝试将cassandra用作键值查找库工作。
我们主要使用数据框，并已脱离RDD API。
而不是与表联接，而是将它们加载到spark或
推动加入cassandra并采取措施避免大笔费用
表扫描，我以为我可以编写一个连接的Spark UDF cassandra一个查找一个键
我还想将结果行转换为案例类对象并返回对象。

根据以下问题的回答，我得到了一些信息。 withSessionDo 重用每个节点上可用的基础JVM级别会话 Spark Cassandra Connector proper usage

val connector = CassandraConnector(sparkConf) // I Know this is serializable.

def lookupKey(connector: CassandraConnector,keyspace: String,table: String): UserDefineFunction = udf((key: String) => {
    connector.withSessionDo(session => {
        val stmt = session.prepare(s"SELECT * FROM $keyspace.$table WHERE key = ?")
        val result = session.execute( stmt.bind(key) )
        MyCaseClass(
           fieldl1 = result.getString(0),fieldl2 = result.getInt(1)
           ...
        )
    }
})

会话无法序列化，因此我们无法在udf外部创建一个会话并将其传递给它，因此我们可以使用映射管理器将行转换为case类实例。使用映射管理器的另一种方法，

def lookupKeyAlt(connector: CassandraConnector,table: String): UserDefineFunction = udf((key: String) => {
    connector.withSessionDo(session => {
        val manager = new MappingManager(session)   // session isn't serializable,so creating one outside and passing to udf is not an option if wf we were willing to do the session management.
        val mapperClass = manager.mapper(classOf[MyCaseClass],keyspace)
        mapperClass.get(key)
    }
})

我是Cassandra的新手，所以请允许我回答几个问题。

这些方法中我是否不了解任何陷阱？
在第二种方法中，我了解到我们在每次调用UDF时都会创建一个新的MappingManager（session）。这是否仍将使用jvm级会话并打开更多会话？在每次调用时实例化MappingManager甚至还正确吗？该会话无法序列化，因此我无法在外部创建会话并将其传递给UDF。
将结果Row转换为Case类对象的其他方法有哪些？
有没有更好的选择来进行这种查找？

解决方法

您正在尝试模拟Spark Cassandra Connector（SCC）的功能，但是您的实现会比SCC慢得多，因为您使用的是同步API，并且在SCC是使用异步API，并并行提取多行数据。

实现所需目标的最佳方法是使用Cassandra优化联接（通常称为“直接联接”）。这种连接始终可用于RDD API，但是很长一段时间才仅在商业版本的连接器中可用于Dataframe API。但是从SCC 2.5.0（released in May 2020th）开始，此功能在开源版本中也可用，因此您可以使用它而不是构建其仿真。通过在配置SparkSession时传递spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions（例如，通过命令行），仅在您enable special Catalyst extensions时执行直接联接。之后，您可以通过全部或部分主键对Cassandra表执行联接，SCC会自动将联接转换为对Cassandra的单个请求，这些请求可以非常有效地执行。您可以通过在联接的数据帧上执行explain来检查是否发生了这种情况，因此应该看到类似的内容（查找字符串 Cassandra Direct Join ）：

scala> joined.explain
== Physical Plan ==
Cassandra Direct Join [pk = id#30,c1 = cc1#32] test.jtest1 - Reading (pk,c1,c2,v) Pushed {}
+- *(1) Project [cast(id#28L as int) AS id#30,cast(id#28L as int) AS cc1#32]
   +- *(1) Range (1,5,step=1,splits=8)

我最近wrote a long blog post解释了如何同时使用Dataframe和RDD API对Cassandra中的数据执行有效的连接-我不想在这里重复：-）

Spark UDF使用Cassandra连接器查找键

如何解决Spark UDF使用Cassandra连接器查找键

解决方法

相关推荐