无法使用 akka 获取 CassandraTableScanRDD

Question

我正在尝试 API 和 Akka 一起休息。入口点之一将从我的 Cassandra 数据库中获取 return 数据。所以我可以像那样得到我的 CassandraTableScanRDD :

val spark = SparkSession
    .builder()
    .appName("Spark Cassandra")
    .config("spark.cores.max", "5")
    .config("spark.sql.warehouse.dir", "/tmp")
    .config("spark.cassandra.connection.host", "localhost")
    .config("spark.cassandra.connection.port", "9042")
    .master("local[*]")
    .getOrCreate()

  val connector = CassandraConnector.apply(spark.sparkContext.getConf)
  val sc = spark.sparkContext
  val temp = sc.cassandraTable("scala_firemen", "firemen")
  
  temp.foreach(println)

使用这段代码，我可以获得我需要的所有数据。但是一旦我添加了我的 Akka 代码，我就不能再 print/access 到数据中了。即使我回滚并删除 Akka 代码，我仍然会编码相同的错误：

[error] (run-main-0) java.lang.ExceptionInInitializerError

[error] java.lang.ExceptionInInitializerError

[error] at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)

[error] at org.apache.spark.rdd.RDD.first(RDD.scala:1367)

[error] at main$.delayedEndpoint$main(Server.scala:34)

[error] at main$delayedInit$body.apply(Server.scala:15)

...

ERROR ContextCleaner: Error in cleaning thread

...

ERROR Utils: uncaught error in thread SparkListenerBus, stopping SparkContext

...

ERROR Utils: throw uncaught fatal error in thread SparkListenerBus

...

当我重新创建一个项目时，代码再次工作，所以我怀疑我需要在删除 Akka 代码后清理一些东西才能再次工作。

Answer 1

RDD 被设计为与 Spark 一起使用，在 StreamingContext 中使用它完全没有意义...

对于您的 Akka 应用程序，您需要 Java driver directly, and even in this case, you don't need to perform full scan of data - fast access to the data is provided only when you have a least partition key... If you still need to perform full scan, then it's better to use something like this - Spark 使用类似的方法进行数据扫描，但您不需要 link 它的所有库。

我建议您了解 Cassandra 的工作原理 - 例如，通过 courses on DataStax Academy - 他们将解释您如何访问数据等。

Answer 2

好的，我发现我不能拥有 "com.typesafe.play" %% "play-json" 与所有 spark 依赖关系的依赖关系，例如：

"org.apache.spark" %% "spark-sql" % "2.1.1",
"org.apache.spark" %% "spark-streaming" % "2.1.1",
"com.datastax.spark" %% "spark-cassandra-connector" % "2.0.10"

无法使用 akka 获取 CassandraTableScanRDD

Can't get CassandraTableScanRDD with akka

scala

apache-spark

apache-spark-sql

akka

spark-cassandra-connector