Convert json to Object in scala using jackson

{
  "sharedObjectName": "sharedObjectName",
  "writeInProgress":false,
  "tableName":"item",
  "lastSyncTimeInEpoch":1619783310,
  "latestLocation":"latestLocation/latestLocation/latestLocation",
  "oneStepLagLocation":"oneStepLagLocation/oneStepLagLocation/oneStepLagLocation"
}

POJO class

case class SharedObject(
  var sharedObjectName: String,
  var writeInProgress: Boolean = false,
  var tableName: String = null,
  var lastSyncTimeInEpoch: Long = 0,
  var latestLocation: String = null,
  var oneStepLagLocation: String = null)

Code to convert the class

import scala.io.Source

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule

def convertJSONToSharedObject( inputMessage : String ): SharedObject ={
  log.info(s"Input message to be converted is $inputMessage")
  try {
    val mapper :ObjectMapper = new ObjectMapper()
    mapper.registerModule(DefaultScalaModule)
    val parsed : SharedObject = mapper.readValue(inputMessage, classOf[SharedObject])
    parsed
  }catch {
    case e : Exception =>
      throw ConfigException(s"Message $inputMessage is not parsable")
  }
}

SBT hacks

Skipping test cases in build

  sbt 'set test in assembly := {}' clean assembly

  sbt 'set test in assembly := {}' clean package

Referring to the local jar in sbt

unmanagedJars in Compile += file("/Users/basan/Documents/Projects/application/application-core/target/scala-2.11/application-core-assembly-1.8.2.jar")

https://www.scala-sbt.org/1.x/docs/Library-Management.html

unmanagedBase := baseDirectory.value / "custom_lib"

https://github.com/sbt/sbt-assembly/blob/master/src/sbt-test/sbt-assembly/deps/build.sbt







lazy val root = (project in file(".")).
	settings(
	version := "0.1",
	scalaVersion := "2.11.12",
	libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.1" % "test",
	libraryDependencies += "ch.qos.logback" % "logback-classic" % "0.9.29" % "runtime",
	unmanagedJars in Compile ++= {
	(baseDirectory.value / "lib" / "compile" ** "*.jar").classpath
	},
	unmanagedJars in Runtime ++= {
	(baseDirectory.value / "lib" / "runtime" ** "*.jar").classpath
	},
	unmanagedJars in Test ++= {
	(baseDirectory.value / "lib" / "test" ** "*.jar").classpath
	},
	assemblyExcludedJars in assembly := {
	(fullClasspath in assembly).value filter {_.data.getName == "compile-0.1.0.jar"}
	},
	assemblyJarName in assembly := "foo.jar",
	TaskKey[Unit]("check") := {
	val process = sys.process.Process("java", Seq("-jar", (crossTarget.value / "foo.jar").toString))
	val out = (process!!)
	if (out.trim != "hello") sys.error("unexpected output: " + out)
	()
	}
	)

Typesafe config map

app {
  #application name
  name: "Streaming"

  dataportal {
        timelinessKafka {
           topicName = "kelsa-DIP-replace"
           producerDetails {
           bootstrap.servers = "broker:9093"
           client.id = "application name"
           key.serializer="org.apache.kafka.common.serialization.StringSerializer"
           value.serializer="org.apache.kafka.common.serialization.StringSerializer"
           ssl.truststore.location = "Data/certificates/metakafka/client.truststore_stg.jks"
           ssl.keystore.location = "Data/certificates/metakafka/server.keystore_stg.jks"
           security.protocol = "SSL"
           ssl.protocol = "TLSv1.2"
           ssl.endpoint.identification.algorithm = ""
           }
        }
    }
}

spark-shell --packages com.typesafe:config:1.3.3

import com.typesafe.config._
import java.io.File
import scala.collection.JavaConverters._


val path = "/Users/basan/Documents/Projects/metricPush2/application.conf"
val config =ConfigFactory.parseFile(new File(path))

val producerDetails = config.getConfig("app.dataportal.timelinessKafka.producerDetails")
producerDetails.getString("bootstrap.servers")


producerDetails
      .entrySet()
      .asScala
      .map( entry => entry.getKey -> entry.getValue.unwrapped().toString ).toMap

res7: scala.collection.immutable.Map[String,String] = Map(ssl.truststore.location -> Data/certificates/metakafka/client.truststore_stg.jks, security.protocol -> SSL, bootstrap.servers -> broker:9093, client.id -> application name, value.serializer -> org.apache.kafka.common.serialization.StringSerializer, ssl.protocol -> TLSv1.2, ssl.endpoint.identification.algorithm -> "", key.serializer -> org.apache.kafka.common.serialization.StringSerializer, ssl.keystore.location -> Data/certificates/metakafka/server.keystore_stg.jks)


config
      .entrySet()
      .asScala
      .map( entry => entry.getKey -> entry.getValue.toString ).toMap

res5: scala.collection.immutable.Map[String,String] = Map(app.dataportal.timelinessKafka.producerDetails.ssl.keystore.location -> Quoted("Data/certificates/metakafka/server.keystore_stg.jks"), app.contactDetails -> Quoted("DataSciences-DE-GSCL-DigitalFulfillment@Target.com"), app.teamName -> Quoted("GSCL-Digital Fulfillment"), app.dataportal.timelinessKafka.producerDetails.key.serializer -> Quoted("org.apache.kafka.common.serialization.StringSerializer"), app.dataportal.timelinessKafka.producerDetails.value.serializer -> Quoted("org.apache.kafka.common.serialization.StringSerializer"), app.dataportal.timelinessKafka.producerDetails.bootstrap.servers -> Quoted("broker:9093"), app.kelsaApplicationPath -> Quoted("Data/KelsaPath/"), app.dataportal.timelinessKafka.producerDetails.client.id -...

Things to consider while buying Residential Plots

1. The site should be proper rectangle/square with the only exception there is an extension in the northeast side. Avoid buying sites with any other shapes.

2. Prefer to buy  40*60feet dimension, if not at least 30*50 feet, Any width higher than 30feet is good though it is difficult to get that kind of plots.

3. Buy only the sites which are having east-facing or north facing.

4. The site should have red soil. Some people might have just poured red soil in the plot for a couple of feet, make sure you dig it for a couple of feet and see.

5. Always try to get history/documents about the plot you are trying to buy and get it verified from a trustworthy/knowledgeable lawyer.

6. If allowed you can check whether you have underground water or not as well, the preferred direction is North-East corner of the plot.

7. It is better to interact directly with the owner while buying plots.

8. Check the map of the area to avoid cremation centers, garbage dump areas, etc.

9. Suggested buy land from the owner in the group, so that you will have the strength to fight any problems.

10. If someone is giving a good offer, if you book the site on the same day then that is the TRAP, always take time to think and do your research.

Spark Concepts:

Spark Concepts:

HashAggregate

Stage
Resource allocation dependents on number of messages/files
we would like to process.

Though even if you allocate 16 cores. if there is file 100KB
then only 1 task will run. Look at the launch time to see
how many threads/tasks are running in parallel.

Always look at Tasks - these are threads.



* is an optimization


Exchange:Shuffle is costly operation. Should avoid as much as possible.
HashPartitioning : matching records will be brought to same machine.
So that each individual can work independently.
Its costly as io is involved,


Smortmergejoin
rabgejoin.

We can givehints.

sparksession.conf.set

Filter data in spark using isin

scala> :paste
// Entering paste mode (ctrl-D to finish)

import java.sql.Date
import org.apache.spark.sql._

    import spark.implicits._
    val inputDF :DataFrame = Seq(
      ("50104044", "order1", "orderline11" , "2020-03", Date.valueOf("2020-03-17")),
      ("50104279", "order2", "orderline21","2020-03", Date.valueOf("2020-03-18")),
      ("50104279", "order22", "orderline221","2020-03", Date.valueOf("2020-03-19")),
      ("50102271", "order3", "orderline31","2020-02", Date.valueOf("2020-02-11")),
      ("50104279", "order4", "orderline41","2020-02", Date.valueOf("2020-02-11")),
      ("50104279", "order5", "orderline51","2020-01", Date.valueOf("2020-01-11"))

    ).toDF("key",
      "order_number",
      "order_line_key",
      "order_mon",
      "order_date")

      val items = List("order1", "order2", "order3")

      val filterdf = inputDF.filter($"order_number".isin(items:_*))
      filterdf.show(false)

// Exiting paste mode, now interpreting.

+--------+------------+--------------+---------+----------+
|key     |order_number|order_line_key|order_mon|order_date|
+--------+------------+--------------+---------+----------+
|50104044|order1      |orderline11   |2020-03  |2020-03-17|
|50104279|order2      |orderline21   |2020-03  |2020-03-18|
|50102271|order3      |orderline31   |2020-02  |2020-02-11|
+--------+------------+--------------+---------+----------+

import java.sql.Date
import org.apache.spark.sql._
import spark.implicits._
inputDF: org.apache.spark.sql.DataFrame = [key: string, order_number: string ... 3 more fields]
items: List[String] = List(order1, order2, order3)
filterdf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [key: string, order_number: string ... 3 more fields]

scala>

Fill Substitute null with values in spark Dataframe

import spark.implicits._

val sourceDF = Seq(

           ("2019-12-09", "11", 1, "L1", "I11", "2018"),

           ("2019-12-09", "11", 2, "L2", "I10", "2018"),

           ("2019-12-09", "11", 3, "L3", "I4", "2018"),

           ("2019-12-09", "11", 4, "L4", "I4", "2015"),

           ("2019-12-09", "11", 5, "L5", "I4", "2019"),

           ("2019-12-09", "11", 6, "L6", null, "2014"),

           ("2019-12-09",  null, 7, "L7", null, "2013")

         ).toDF("date", "hour", "order", "line", "item", "time")

sourceDF.show(false)

val filledDF = sourceDF.na.fill("-99999",Seq("hour","item"))

filledDF.show(false)

scala> sourceDF.show(false)

+----------+----+-----+----+----+----+

|date      |hour|order|line|item|time|

+----------+----+-----+----+----+----+

|2019-12-09|11  |1    |L1  |I11 |2018|

|2019-12-09|11  |2    |L2  |I10 |2018|

|2019-12-09|11  |3    |L3  |I4  |2018|

|2019-12-09|11  |4    |L4  |I4  |2015|

|2019-12-09|11  |5    |L5  |I4  |2019|

|2019-12-09|11  |6    |L6  |null|2014|

|2019-12-09|null|7    |L7  |null|2013|

+----------+----+-----+----+----+----+

scala> val filledDF = sourceDF.na.fill("-99999",Seq("hour","item"))

filledDF: org.apache.spark.sql.DataFrame = [date: string, hour: string ... 4 more fields]

scala> filledDF.show(false)

+----------+------+-----+----+------+----+

|date      |hour  |order|line|item  |time|

+----------+------+-----+----+------+----+

|2019-12-09|11    |1    |L1  |I11   |2018|

|2019-12-09|11    |2    |L2  |I10   |2018|

|2019-12-09|11    |3    |L3  |I4    |2018|

|2019-12-09|11    |4    |L4  |I4    |2015|

|2019-12-09|11    |5    |L5  |I4    |2019|

|2019-12-09|11    |6    |L6  |-99999|2014|

|2019-12-09|-99999|7    |L7  |-99999|2013|

+----------+------+-----+----+------+----+

scala>