Converting DataSet to Json Array Spark using Scala

Question

I am new to the spark and unable to figure out the solution for the following problem.

I have a JSON file to parse and then create a couple of metrics and write the data back into the JSON format.

now following is my code I am using

import org.apache.spark.sql._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions._

object quick2 {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession
      .builder
      .appName("quick1")
      .master("local[*]")
      .getOrCreate()

    val rawData = spark.read.json("/home/umesh/Documents/Demo2/src/main/resources/sampleQuick.json")

    val mat1 = rawData.select(rawData("mal_name"),rawData("cust_id")).distinct().orderBy("cust_id").toJSON.cache()
    val mat2 = rawData.select(rawData("file_md5"),rawData("mal_name")).distinct().orderBy(asc("file_md5")).toJSON.cache()

val write1 = mat1.coalesce(1).toJavaRDD.saveAsTextFile("/home/umesh/Documents/Demo2/src/test/mat1/")

val write = mat2.coalesce(1).toJavaRDD.saveAsTextFile("/home/umesh/Documents/Demo2/src/test/mat2/")
}
}

Now above code is writing the proper json format. However, matrices can contain duplicate result as well example:

md5   mal_name
1       a
1       b
2       c
3       d
3       e

so with above code every object is getting written in single line

like this

{"file_md5":"1","mal_name":"a"}
{"file_md5":"1","mal_name":"b"}
{"file_md5":"2","mal_name":"c"}
{"file_md5":"3","mal_name":"d"}

and so on.

but I want to combine the data of common keys:

so the output should be

{"file_md5":"1","mal_name":["a","b"]}

can somebody please suggest me what shall I do here. Or if there is any other better way to approach this problem.

Thanks!

mrsrinivas · Accepted Answer · 2017-07-24 16:12:21Z

5

You can use collect_list or collect_set as per your need on mal_name column

You can directly save DataFrame/DataSet directly as JSON file

import org.apache.spark.sql.functions.{alias, collect_list}
import spark.implicits._

rawData.groupBy($"file_md5")
  .agg(collect_set($"mal_name").alias("mal_name"))
  .write
  .format("json")
  .save("json/file/location/to/save")

edited Jul 24, 2017 at 16:12

answered Jul 24, 2017 at 12:43

mrsrinivas

35.6k13 gold badges133 silver badges132 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Umesh Kaushik · Accepted Answer · 2017-07-24 13:41:38Z

0

as wrote by @mrsrinivas I changed my code as per below

val mat2 = rawData.select(rawData("file_md5"),rawData("mal_name")).distinct().orderBy(asc("file_md5")).cache()
val labeledDf = mat2.toDF("file_md5","mal_name")
labeledDf.groupBy($"file_md5").agg(collect_list($"mal_name")).coalesce(1).write.format("json").save("/home/umesh/Documents/Demo2/src/test/run8/")

Keeping this quesion open for some more suggestions if any.

answered Jul 24, 2017 at 13:41

Umesh Kaushik

1,6191 gold badge18 silver badges18 bronze badges

2 Comments

philantrovert Over a year ago

.agg(collect_list($"mal_name").alias("mal_name")) should save you from creating a new column.

Umesh Kaushik Over a year ago

hi, tried alias did not work gave an error and when I tried with "as" it worked. val labeledDf = mat1.groupBy($"file_md5").agg(collect_set($"mal_name").as("mal_name"))

Collectives™ on Stack Overflow

Converting DataSet to Json Array Spark using Scala

2 Answers 2

Comments

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related