Special characters in spark columns are supported

 

val spark = SparkSession.builder

.master("local[1]")

.appName("SparkByExamples.com")

.getOrCreate()

val data = Seq(("James","Smith","USA","CA"),

("Michael","Rose","USA","NY"),

("Robert","Williams","USA","CA"),

("Maria","Jones","USA","FL")

)

val columns = Seq("firstname","lastname","country","state")

import spark.implicits._

val df = data.toDF(columns:_*)

df.show(false)

df.select("firstname","lastname").show()

================

val spark = SparkSession.builder

.master("local[1]")

.appName("SparkByExamples.com")

.getOrCreate()

val data = Seq(("James","Smith","USA","CA"),

("Michael","Rose","USA","NY"),

("Robert","Williams","USA","CA"),

("Maria","Jones","USA","FL")

)

val columns = Seq("first name","last # name","coun ? try","sta te")

import spark.implicits._

val df = data.toDF(columns:_*)

df.show(false)

df.select("first name","last # name" , "coun ? try").show()

scala> df.select("first name","last # name" , "coun ? try").show() +----------+-----------+----------+ |first name|last # name|coun ? try| +----------+-----------+----------+ | James| Smith| USA| | Michael| Rose| USA| | Robert| Williams| USA| | Maria| Jones| USA| +----------+-----------+----------+

Write the json file to hdfs with spark

 

Pass the valid hdfs path. It will create the single file in overwrite mode

@throws[WriterException]
def writeFileToHDFS(fileLocation: String, fileBody : String)(sparkSession: SparkSession) = {
  logInfo(s"Write the contents to the file location $fileLocation")
  var fsDataOutputStream  : FSDataOutputStream = null
  try {
    val hadoopConf: Configuration = sparkSession.sparkContext.hadoopConfiguration
    val hdfs: FileSystem          = org.apache.hadoop.fs.FileSystem.get(hadoopConf)
    val exportToFile = new Path(fileLocation)

    fsDataOutputStream  = hdfs.create(exportToFile)
    fsDataOutputStream.write(fileBody.getBytes)
    fsDataOutputStream.hflush()
  }catch {
    case ex : Exception => logError(s"Error in writing to the file $fileLocation", ex)
      throw WriterException(s"Error in writing to the file $fileLocation")
  }finally {
    logInfo("Closing the writer object")
    if(fsDataOutputStream != null){
      fsDataOutputStream.close()
    }
  }
}

If we write the dataframe it will create the folder and within that it will write the file

def writeFileToHDFSUsingDataFrame(fileLocation: String, fileBody : String)(sparkSession: SparkSession): Unit ={
  import sparkSession.implicits._
  Seq(fileBody)
    .toDF().coalesce(1)
    .write
    .mode("overwrite").text(fileLocation)
}

Spark Read content of the file as is

 json file

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

Spark Code to read the file

@throws[ConfigException]
def readJSONFileAsIs( location: String)(sparkSession:SparkSession): String ={
  logInfo(s"Read the file contents as is from the location $location")
  try {
    val textFileDataFrame : Dataset[String] = sparkSession.read.textFile(location)
    val builder : StringBuilder = new StringBuilder()
    textFileDataFrame.collect().foreach(line=>{
      builder.append(line)
    })
    val fileContents : String = builder.toString()
    logInfo(s"Content of the file $fileContents")
    fileContents

  }catch {
    case exception: Exception => logError(s"Error in reading the file from the location $location",exception)
      throw  ConfigException(s"Error in reading the file from the location $location")
  }

}

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.