将数据从 Spark-Streaming 存储到 Cassandra 时出现问题答案

【问题标题】：Issue while storing data from Spark-Streaming to Cassandra将数据从 Spark-Streaming 存储到 Cassandra 时出现问题
【发布时间】：2017-01-14 18:31:36
【问题描述】：

SparkStreaming 上下文从 RabbitMQ 以 30 秒的间隔读取流。我想修改 cassandra 中存在的相应行的几列的值，然后想将数据存储回Cassandra。为此，我需要检查特定主键的行是否存在于 Cassandra 中，如果是，请获取它并执行必要的操作。但问题是，我在驱动程序上创建了StreamingContext，并在 Worker 上执行了操作。因此，他们无法获得 StreamingContext 对象，原因是它没有被序列化并发送给工作人员，我收到了这个错误： java.io.NotSerializableException: org.apache.spark.streaming.StreamingContext。我也知道我们无法访问 foreachRDD 中的StreamingContext。但是，我如何在这里实现相同的功能而不会出现序列化错误？

我看过几个例子here，但没有帮助。

这里是sn-p的代码：

   val ssc = new StreamingContext(sparkConf,30)
    val receiverStream = RabbitMQUtils.createStream(ssc, rabbitParams)
    receiverStream.start()      
    val lines = receiverStream.map(EventData.fromString(_))
    lines.foreachRDD{ x => if (x.toLocalIterator.nonEmpty) {
                x.foreachPartition { it => for (tuple <- it) { 
                val cookieid  = tuple.cookieid                
                val sessionid = tuple.sessionid              
                val logdate = tuple.logdate
                val EventRows =  ssc.cassandraTable("SparkTest", CassandraTable).select("*")
                .where("cookieid = '" + cookieid + "' and logdate = '" + logdate+ "' and sessionid = '" + sessionid + "')

                   Somelogic Whether row exist or not for Cookieid

                }  } }

【问题讨论】：

how to create a new RDD in anonymous function in spark streaming?的可能重复
@eliasah 我无法在 lines.foreachRDD 内访问 StreamingContext，无论您在外部还是内部声明 StreamingContext。你为什么把它标记为关闭？它没有解决我的问题，仅供参考，它不是您提供的链接的副本。
从x.sparkContext()获取火花上下文，不要从forEachMethod调用

标签： apache-spark serialization cassandra spark-streaming

【解决方案1】：

SparkContext 不能被序列化并在可能不同节点中的多个工作人员之间传递。如果你需要做这样的事情，你可以使用 forEachPartiion、mapPartitons。否则，请使用您传递的函数来执行此操作

 CassandraConnector(SparkWriter.conf).withSessionDo { session =>
  ....
    session.executeAsync(<CQL Statement>)

在 SparkConf 中，您需要提供 Cassandra 的详细信息

  val conf = new SparkConf()
    .setAppName("test")
    .set("spark.ui.enabled", "true")
    .set("spark.executor.memory", "8g")
    //  .set("spark.executor.core", "4")
    .set("spark.eventLog.enabled", "true")
    .set("spark.eventLog.dir", "/ephemeral/spark-events")
    //to avoid disk space issues - default is /tmp
    .set("spark.local.dir", "/ephemeral/spark-scratch")
    .set("spark.cleaner.ttl", "10000")
    .set("spark.cassandra.connection.host", cassandraip)
    .setMaster("spark://10.255.49.238:7077")

Java CSParser 是一个不可序列化的库。因此，如果您在 RDD 上调用 map 或 forEach，Spark 将无法向其发送可能不同的节点。一种解决方法是使用 mapPartion，在这种情况下，一个完整的分区将在一个 SparkNode 中执行。因此它不需要为每个调用序列化。示例

val rdd_inital_parse = rdd.mapPartitions(pLines).

 def pLines(lines: Iterator[String]) = {
    val parser = new CSVParser() ---> Cannot be serialized, will fail if using rdd.map(pLines)
    lines.map(x => parseCSVLine(x, parser.parseLine))
  }

【讨论】：

如果我使用lines.foreachRDD{ x => if (x.toLocalIterator.nonEmpty) { x.foreachPartition { CassandraConnector(ssc.sparkContext.getConf).withSessionDo { session => .... session.executeAsync(<CQL Statement>) ssc，worker 将无法使用。所以，这又是我问你的同一个问题

【解决方案2】：

尝试使用 x.sparkContext.cassandraTable() 而不是 ssc.cassandraTable() 看看是否有帮助

【讨论】：

它不起作用。并请在 cmets 中写下建议
这确实有帮助！