Generate JSON with Null Columns in Spark

Generate JSON with Null Columns in Spark

Install JDK8 Oracle on Ubuntu 18.04
https://www.digitalocean.com/community/tutorials/how-to-install-java-with-apt-on-ubuntu-18-04
> sudo add-apt-repository ppa:webupd8team/java
> sudo apt update
> sudo apt install oracle-java8-installer
Java Version
> java -version
java version "1.8.0_201"
Java(TM) SE Runtime Environment (build 1.8.0_201-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.201-b09, mixed mode)

Having Hadoop and Zeppelin Running
Dump the CSV data from DynamoDB
> export-dynamodb -t meeting-prod-meetings -f csv -o meetingdb.csv
Create the data directory
> hdfs dfs -mkdir hdfs://localhost:9000/meetingpin
Put the data to remote
> hdfs dfs -put ./meetingdb.csv hdfs://localhost:9000/meetingpin/meetingdb.csv
Fetch the result to local
> hdfs dfs -get hdfs://localhost:9000/meetingpin/meetings_to_updatepin ./meetings_to_updatepin
Here are some normal steps to investigate the data
val meetingRawDF = sqlContext.read.format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("hdfs://localhost:9000/meetingpin/meetingdb.csv") 
val meeting1DF = meetingRawDF.toDF(meetingRawDF.columns map(_.toLowerCase): _*)
val meetingDF = meeting1DF.columns.foldLeft(meeting1DF)((curr, n) => curr.withColumnRenamed(n, n.replaceAll("\\s", "_")))
meetingDF.printSchema()
meetingDF.createOrReplaceTempView("meetings")
%sql
SELECT
    pinrequired as pinRequired,
    maxparticipants as maxParticipants,
    lastcalldate as lastCallDate,
    clientapiid as clientApiID,
    ownerclientapiid as ownerClientApiID,
    uuid as UUID,
    displayname as displayName,
    email as email
FROM meetings
WHERE pin = 'NA' limit 5
Here is the hard part, because usually, the normal thing is to ignore the JSON key if the value is null, we need to create a Case Class and doing the converting to JSON ourselves
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper
import spark.implicits._
case class MeetingToUpdate(UUID:String, pin:String, action:String, `type`:String)
val meetingsToUpdateDF = sqlContext.sql("""
SELECT
    uuid as UUID,
    null as pin,
    'UPDATED' as action,
    'MEETING' as type
FROM meetings
WHERE pin = 'NA'
""")
val meetingsCaseDF = meetingsToUpdateDF.map(row => MeetingToUpdate(row.getString(0), null, row.getString(2), row.getString(3)))
val meetingsToUpdateNullDF = meetingsCaseDF.mapPartitions(records => {
    val mapper = new ObjectMapper with ScalaObjectMapper
        mapper.registerModule(DefaultScalaModule)
        records.map(mapper.writeValueAsString(_))
    }
)

meetingsToUpdateNullDF.show(1)
meetingsToUpdateNullDF.coalesce(1).write.text("hdfs://localhost:9000/meetingpin/meetings_to_updatepin2")

References:
https://stackoverflow.com/questions/28166555/how-to-convert-row-of-a-scala-dataframe-into-case-class-most-efficiently
https://stackoverflow.com/questions/45235593/write-a-spark-dataset-to-json-with-all-keys-in-the-schema-including-null-column?rq=1
https://stackoverflow.com/questions/44271612/retain-keys-with-null-values-while-writing-json-in-spark