suddenly throwing This RDD lacks a SparkContext it

问题:

It was a working piece of code but suddenly its not working after I tried creating Sparksession from different scala object

val b = a.filter { x => (!x._2._1.isEmpty) && (!x._2._2.isEmpty) } 

val primary_ke = b.map(rec => (rec._1.split(",")(0))).distinct 

for (i <- primary_key_distinct) {    
  b.foreach(println)

}

Error:

 ERROR Executor: Exception in task 0.0 in stage 5.0 (TID 5)
org.apache.spark.SparkException: This RDD lacks a SparkContext. It could happen in the following cases: 
(1) RDD transformations and actions are NOT invoked by the driver, but inside of other transformations; for example, rdd1.map(x => rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063.
(2) When a Spark Streaming job recovers from checkpoint, this exception will be hit if a reference to an RDD not defined by the streaming job is used in DStream operations. For more information, See SPARK-13758.

Not working even after I revoked it and I'm not using any objects.

Code Updated:

object try {

  def main(args: Array[String]) {



val spark = SparkSession.builder().master("local").appName("50columns3nodes").getOrCreate()

var s = spark.read.csv("/home/hadoopuser/Desktop/input/source.csv").rdd.map(_.mkString(","))
var k = spark.read.csv("/home/hadoopuser/Desktop/input/destination.csv").rdd.map(_.mkString(","))

val source_primary_key = s.map(rec => (rec.split(",")(0), rec))
val destination_primary_key = k.map(rec => (rec.split(",")(0), rec))

val a = source_primary_key.cogroup(destination_primary_key).filter { x => ((x._2._1) != (x._2._2)) }
val b = a.filter { x => (!x._2._1.isEmpty) && (!x._2._2.isEmpty) } 

var extra_In_Dest = a.filter(x => x._2._1.isEmpty && !x._2._2.isEmpty).map(rec => (rec._2._2.mkString(""))) 
var extra_In_Src = a.filter(x => !x._2._1.isEmpty && x._2._2.isEmpty).map(rec => (rec._2._1.mkString(""))) 

val primary_key_distinct = b.map(rec => (rec._1.split(",")(0))).distinct 
for (i <- primary_key_distinct) {

  var lengthofarray = 0
  println(i)
  b.foreach(println)

}
}
}

Input data follows

s=1,david 2,ajay 3,jijo 4,abi 5,surendhar

k=1,david 2,ajay 3,jijoaa 4,abisdsdd 5,surendhar

val a contains {3,(jijo,jijoaa),5(abi,abisdsdd)}

回答1:

If you read carefully the first message

(1) RDD transformations and actions are NOT invoked by the driver, but inside of other transformations; for example, rdd1.map(x => rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063.

It clearly states that actions and transformations cannot be performed inside a transformation.

primary_key_distinct is transformation done on b and b itself is a transformation done on a. And b.foreach(println) is an action done inside transformation of primary_key_distinct

So if you collect b or primary_key_distinct inside driver, then the code should run properly

val b = a.filter { x => (!x._2._1.isEmpty) && (!x._2._2.isEmpty) }.collect

or

val primary_key_distinct = b.map(rec => (rec._1.split(",")(0))).distinct.collect

or if you don't use action inside another transformation then the code should run properly too as

for (i <- 1 to 2) {

  var lengthofarray = 0
  println(i)
  b.foreach(println)

}

I hope the explanation is clear.