ISO 8601 format to Date

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

import org.apache.spark.sql.functions._

import org.apache.spark.sql.Column;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;

import org.apache.spark.sql.types.DataType;

import org.apache.spark.sql.types.DataTypes;

    val dfDate = Seq(("2021-01-25T13:33:44.343Z"),

    ("2019-02-05T14:06:31.556+0100"),

    ("2021-01-25T13:33:44.343+1:00")).toDF("input_timestamp")

    dfDate.show(false)

println("====   apply the to_timestamp method, loosing sub seconds=====")

  val resultdf = dfDate.withColumn("datetype_timestamp",to_timestamp(col("input_timestamp"),"yyyy-MM-dd'T'HH:mm:ss.SSSZ"))

  resultdf.printSchema

  resultdf.show(false)

println("==== explicit cast to timestamp... loosing sub seconds=====")

   val resultdf = dfDate.withColumn("datetype_timestamp",to_timestamp(col("input_timestamp"),"yyyy-MM-dd'T'HH:mm:ss.SSSZ").cast(TimestampType))

  resultdf.printSchema

  resultdf.show(false)

println("====  cast to timestamp... retains sub seconds  =====")

  val resultdf = dfDate.withColumn("datetype_timestamp",col("input_timestamp").cast(TimestampType))

   resultdf.printSchema

  resultdf.show(false)

println("====  cast to date... retains sub seconds  =====")

  val resultdf = dfDate.withColumn("datetype_timestamp",to_date(col("input_timestamp"),"yyyy-MM-dd'T'HH:mm:ss.SSSZ"))

   resultdf.printSchema

  resultdf.show(false)

println("====  parse as String  =====")

  val resultdf = dfDate.withColumn("datetype_timestamp1",date_format(col("input_timestamp"),"yyyy-MM-dd'T'HH:mm:ss.SSSZ"))

 resultdf.printSchema

  resultdf.show(false)

println("====  parsed  String cast it  =====")

  val resultdf3 = resultdf.withColumn("datetype_timestamp2",col("datetype_timestamp1").cast(TimestampType))

 resultdf3.printSchema

  resultdf3.show(false)

Parsing ISO 8601 format dates in spark

 

import org.apache.spark.sql.functions._

import org.apache.spark.sql.Column;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;

import org.apache.spark.sql.types.DataType;

import org.apache.spark.sql.types.DataTypes;

    val dfDate = Seq(("2021-01-25T13:33:44.343Z"),

    ("2019-02-05T14:06:31.556+0100")).toDF("input_timestamp")

  dfDate.withColumn("datetype_timestamp",to_timestamp(col("input_timestamp"),"yyyy-MM-dd'T'HH:mm:ss.SSSZ")).show(false)

  

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

|input_timestamp             |datetype_timestamp |

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

|2021-01-25T13:33:44.343Z    |null               |

|2019-02-05T14:06:31.556+0100|2019-02-05 18:36:31|

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

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)
	()
	}
	)