String filter using Spark UDF

Question

input.csv:

200,300,889,767,9908,7768,9090

300,400,223,4456,3214,6675,333

234,567,890

123,445,667,887

What I want: Read input file and compare with set "123,200,300" if match found, gives matching data 200,300 (from 1 input line)

300 (from 2 input line)

123 (from 4 input line)

What I wrote:

  import org.apache.spark.{SparkConf, SparkContext}
  import org.apache.spark.rdd.RDD

  object sparkApp {
    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("CountingSheep")
    val sc = new SparkContext(conf)

    def parseLine(invCol: String) : RDD[String]  = {
      println(s"INPUT, $invCol")
      val inv_rdd = sc.parallelize(Seq(invCol.toString))
      val bs_meta_rdd = sc.parallelize(Seq("123,200,300"))
      return inv_rdd.intersection(bs_meta_rdd)
    }

    def main(args: Array[String]) {
      val filePathName = "hdfs://xxx/tmp/input.csv"
      val rawData = sc.textFile(filePathName)
      val datad = rawData.map{r => parseLine(r)}
    }
  }

I get the following exception:

java.lang.NullPointerException

Please suggest where I went wrong

Answer 1

Problem is solved. This is very simple.

val pfile = sc.textFile("/FileStore/tables/6mjxi2uz1492576337920/input.csv")
case class pSchema(id: Int, pName: String)
val pDF = pfile.map(_.split("\t")).map(p => pSchema(p(0).toInt,p(1).trim())).toDF()
pDF.select("id","pName").show()

Define UDF

val findP = udf((id: Int,
                    pName: String
                    ) => {
  val ids = Array("123","200","300")
  var idsFound : String = ""
  for (id  <- ids){
    if (pName.contains(id)){
      idsFound = idsFound + id + ","
    }
  }
  if (idsFound.length() > 0) {
    idsFound = idsFound.substring(0,idsFound.length -1)
  }    
 idsFound
})

Use UDF in withCoulmn()

pDF.select("id","pName").withColumn("Found",findP($"id",$"pName")).show()

Answer 2

What you are trying to do can't be done the way you are doing it.

Spark does not support nested RDDs (see SPARK-5063).

Spark does not support nested RDDs or performing Spark actions inside of transformations; this usually leads to NullPointerExceptions (see SPARK-718 as one example). The confusing NPE is one of the most common sources of Spark questions on StackOverflow:

• call of distinct and map together throws NPE in spark library

• NullPointerException in Scala Spark, appears to be caused be collection type?

• Graphx: I've got NullPointerException inside mapVertices (those are just a sample of the ones that I've answered personally; there are many others).

I think we can detect these errors by adding logic to RDD to check whether sc is null (e.g. turn sc into a getter function); we can use this to add a better error message.

Answer 3

For simple answer, why we are making it so complex? In this case we don't require UDF.

This is your input data:

200,300,889,767,9908,7768,9090|AAA
300,400,223,4456,3214,6675,333|BBB
234,567,890|CCC
123,445,667,887|DDD

and you have to match it with 123,200,300

val matchSet = "123,200,300".split(",").toSet
val rawrdd = sc.textFile("D:\\input.txt")
rawrdd.map(_.split("|"))
      .map(arr => arr(0).split(",").toSet.intersect(matchSet).mkString(",") + "|" + arr(1))
      .foreach(println)

Your output:

300,200|AAA
300|BBB
|CCC
123|DDD