Spark Unit Test First example using scalatest

package com.chapter16.SparkTesting
import org.scalatest.{ BeforeAndAfterAll, FunSuite }
import org.scalatest.Assertions._
import org.apache.spark.sql.SparkSession
import org.apache.spark.rdd.RDD
class wordCountTest2 extends FunSuite with BeforeAndAfterAll {
  var spark: SparkSession = null
  def tokenize(line: RDD[String]) = {
    line.map(x => x.split(' ')).collect()
  }
  override def beforeAll() {
    spark = SparkSession
      .builder
      .master("local[*]")
      .config("spark.sql.warehouse.dir", "E:/Exp/")
      .appName(s"OneVsRestExample")
      .getOrCreate()
  }  
  test("Test if two RDDs are equal") {
    val input = List("To be,", "or not to be:", "that is the question-", "William Shakespeare")
    val expected = Array(Array("To", "be,"), Array("or", "not", "to", "be:"), Array("that", "is", "the", "question-"), Array("William", "Shakespeare"))
    val transformed = tokenize(spark.sparkContext.parallelize(input))
    assert(transformed === expected)
  }  
  test("Test for word count RDD") {
    val fileName = "C:/Users/rezkar/Downloads/words.txt"
    val obj = new wordCountRDD
    val result = obj.prepareWordCountRDD(fileName, spark)    
    assert(result.count() === 214)
  }
  override def afterAll() {
    spark.stop()
  }
}

Creating Dataframe of structtype

val dfLocal2 = Seq(
      (1,"PRESCHOOL", true, "basan", "11", "asdas", "3", "LINES"),
      (2, null,        false, "basan", "32", "asfasf", "13", "/COSMETICS"),
      (3,"FURNITURE",         null, "basan", "23", "asfasf HOME", "4", "HOME")

    ).toDF("id_int_null","string_null","boolean_null","TCIN","DIVISION_ID","DIVISION_NAME","GROUP_ID","GROUP_NAME")
dfLocal2.show()
dfLocal2.printSchema()



___


import org.apache.spark.sql.Row
import org.apache.spark.sql._
import org.apache.spark.sql.types._

val someData = Seq(
  Row(8, "bat"),
  Row(64, "mouse"),
  Row(-27, "horse"),
  Row(null,"abc")
)

val someSchema = List(
  StructField("number", IntegerType, true),
  StructField("word", StringType, true)
)

val someDF = spark.createDataFrame(
  sc.parallelize(someData),
  StructType(someSchema)
)

someDF.show(6)

someDF.show()

Try Success failure on RDD operation

import scala.util.{Failure, Success, Try}
import org.apache.spark.sql.Row
     
     
      val rdd= spark.sparkContext.parallelize(Seq(("Java", 20000), ("Python", 100000), ("Scala", 3000)))

        val goodBadRecords = rdd.map(line =>
          Try{
            print(line)
         
          } match {
            case Success(map) => Right(map)
            case Failure(e) => Left(e)
          }

        )

        val records = goodBadRecords.filter(_.isRight)
        val errors = goodBadRecords.filter(_.isLeft)
       
       
        records.count()
        errors.count()

create Dstream manually using REPL

import scala.collection.mutable
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
val lines = Seq("To be or not to be.", "That is the question.")

val rdd = sc.parallelize(lines)

val lines = mutable.Queue[RDD[String]]()

val streamingContext = new StreamingContext(sc, Seconds(10))
val dstream = streamingContext.queueStream(lines)

// append data to DStream
lines += sc.makeRDD(Seq("To be or not to be.", "That is the question."))


http://mkuthan.github.io/blog/2015/03/01/spark-unit-testing/

Add fixed value column in dataframe

scala> val dfLocal = Seq(
     |              ("DEPARTMENT_ID1","DEPARTMENT_NAME1","GROUP_NAME1"),
     |
     |               ("1","PRESCHOOL",      "asdasf"),
     |               ("2","COSMETICS",              "COSMETICS"),
     |               ("3","FURNITURE",           "HOME")
     |             ).toDF("DEPARTMENT_ID","DEPARTMENT_NAME","GROUP_NAME")
dfLocal: org.apache.spark.sql.DataFrame = [DEPARTMENT_ID: string, DEPARTMENT_NAME: string ... 1 more field]

scala> dfLocal.printSchema()
root
 |-- DEPARTMENT_ID: string (nullable = true)
 |-- DEPARTMENT_NAME: string (nullable = true)
 |-- GROUP_NAME: string (nullable = true)


scala> import org.apache.spark.sql.functions.typedLit
import org.apache.spark.sql.functions.typedLit

scala> val successDataSet2 = dfLocal.withColumn("newcolumnadded", lit("204"))
successDataSet2: org.apache.spark.sql.DataFrame = [DEPARTMENT_ID: string, DEPARTMENT_NAME: string ... 2 more fields]

scala> successDataSet2.show(2)
+--------------+----------------+-----------+--------------+
| DEPARTMENT_ID| DEPARTMENT_NAME| GROUP_NAME|newcolumnadded|
+--------------+----------------+-----------+--------------+
|DEPARTMENT_ID1|DEPARTMENT_NAME1|GROUP_NAME1|           204|
|             1|       PRESCHOOL|     asdasf|           204|
+--------------+----------------+-----------+--------------+
only showing top 2 rows


scala> successDataSet2.printSchema()
root
 |-- DEPARTMENT_ID: string (nullable = true)
 |-- DEPARTMENT_NAME: string (nullable = true)
 |-- GROUP_NAME: string (nullable = true)
 |-- newcolumnadded: string (nullable = false)


scala>

De dupe in Dataframe

val df = Seq(
  ("d1", "2018-09-20 10:00:00", "blah1"),
  ("d2", "2018-09-20 09:00:00", "blah2"),
  ("d1", "2018-09-20 10:01:00", "blahnew")
).toDF("documentId","timestamp","anotherField")

import org.apache.spark.sql.functions.row_number
import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"documentId").orderBy($"timestamp".desc)
val Resultdf = df.withColumn("rownum", row_number.over(w))
     .where($"rownum" === 1).drop("rownum")

Resultdf.show()
input:

+----------+-------------------+------------+
|documentId|          timestamp|anotherField|
+----------+-------------------+------------+
|        d1|2018-09-20 10:00:00|       blah1|
|        d2|2018-09-20 09:00:00|       blah2|
|        d1|2018-09-20 10:01:00|     blahnew|
+----------+-------------------+------------+
output:

+----------+-------------------+------------+
|documentId|          timestamp|anotherField|
+----------+-------------------+------------+
|        d2|2018-09-20 09:00:00|       blah2|
|        d1|2018-09-20 10:01:00|     blahnew|
+----------+-------------------+------------+

Converting string date value to Date object

scala> import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.IntegerType

scala> import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StringType

scala> import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructType

scala> import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StructField

scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row

scala> import scala.collection.JavaConversions._
import scala.collection.JavaConversions._

scala>     val data = Seq(("Java", 20000 , "2012-02-05"), ("Python", 100000 , "2012-02-06"), ("Scala", 3000 , "2012-02-07"))
data: Seq[(String, Int, String)] = List((Java,20000,2012-02-05), (Python,100000,2012-02-06), (Scala,3000,2012-02-07))

scala>     val tableColumns = List(("language", "string"), ("people", "integer"), ("registerdate", "String"))
tableColumns: List[(String, String)] = List((language,string), (people,integer), (registerdate,String))

scala>     var schema = new StructType
schema: org.apache.spark.sql.types.StructType = StructType()

scala>     for (i <- p="" tablecolumns="">     |       schema = schema.add(i._1, i._2)

scala>       val rowData = data.map(attributes => Row(attributes._1, attributes._2 ,  attributes._3))
rowData: Seq[org.apache.spark.sql.Row] = List([Java,20000,2012-02-05], [Python,100000,2012-02-06], [Scala,3000,2012-02-07])

scala>       var dfFromData4 = spark.createDataFrame(rowData,schema)
dfFromData4: org.apache.spark.sql.DataFrame = [language: string, people: int ... 1 more field]

scala>       dfFromData4.show(2)
+--------+------+------------+
|language|people|registerdate|
+--------+------+------------+
|    Java| 20000|  2012-02-05|
|  Python|100000|  2012-02-06|
+--------+------+------------+
only showing top 2 rows


scala>

scala>         import org.apache.spark.sql.functions.to_date
import org.apache.spark.sql.functions.to_date

scala>         val dateFormat = "yyyy-dd-MM"
dateFormat: String = yyyy-dd-MM

scala>        var dfFromData5 = dfFromData4.withColumn("registerdate2", to_date(col("registerdate"), dateFormat))
dfFromData5: org.apache.spark.sql.DataFrame = [language: string, people: int ... 2 more fields]

scala>        dfFromData5.show(3)
+--------+------+------------+-------------+
|language|people|registerdate|registerdate2|
+--------+------+------------+-------------+
|    Java| 20000|  2012-02-05|   2012-05-02|
|  Python|100000|  2012-02-06|   2012-06-02|
|   Scala|  3000|  2012-02-07|   2012-07-02|
+--------+------+------------+-------------+


scala> dfFromData5.printSchema()
root
 |-- language: string (nullable = true)
 |-- people: integer (nullable = true)
 |-- registerdate: string (nullable = true)
 |-- registerdate2: date (nullable = true)