Apache Spark Read JSON With Extra Columns

Question

I'm reading a Hive table which has two columns, id and jsonString. I can easily transform the jsonString into a Spark Data Structure calling the spark.read.json function, but I have to add the column id as well.

val jsonStr1 = """{"fruits":[{"fruit":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""
val jsonStr2 = """{"fruits":[{"dt":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""
val jsonStr3 = """{"fruits":[{"a":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""


case class Foo(id: Integer, json: String)

val ds = Seq(new Foo(1,jsonStr1), new Foo(2,jsonStr2), new Foo(3,jsonStr3)).toDS
val jsonDF = spark.read.json(ds.select($"json").rdd.map(r => r.getAs[String](0)).toDS)

jsonDF.show()

jsonDF.show
+--------------------+------------------+------------------+--------------------+
|                 bar|              cars|            daniel|              fruits|
+--------------------+------------------+------------------+--------------------+
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|
+--------------------+------------------+------------------+--------------------+

I would like to add the column id from the Hive table, like this:

+--------------------+------------------+------------------+--------------------+---------------
|                 bar|              cars|            daniel|              fruits|  id
+--------------------+------------------+------------------+--------------------+--------------
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|1
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|2
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|3
+--------------------+------------------+------------------+--------------------+

I will not use regular expressions

I created a udf which take this two fields as argument and using a proper JSON library include the desired field(id) and return a new JSON string. It works like a charm but I hope Spark API offers a better way to do it. I'm using Apache Spark 2.3.0.

Stornu2 · Accepted Answer · 2020-02-12 12:35:12Z

I already knew about the from_json function before, but in my case it would be "impossible" to manually infer the schema for each JSON. I was thinking that Spark would have an "idiomatic" interface.

This is my final solution:

ds.select($"id", from_json($"json", jsonDF.schema).alias("_json_path")).select($"_json_path.*", $"id").show

ds.select($"id", from_json($"json", jsonDF.schema).alias("_json_path")).select($"_json_path.*", $"id").show

+--------------------+------------------+------------------+--------------------+---+
|                 bar|              cars|            daniel|              fruits| id|
+--------------------+------------------+------------------+--------------------+---+
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|  1|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|  2|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|  3|
+--------------------+------------------+------------------+--------------------+---+

Leo C · Accepted Answer · 2019-03-09 07:38:05Z

2

One way would be to apply from_json to the JSON strings with the corresponding schema, as shown below:

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

case class Foo(id: Int, json: String)

val df = Seq(Foo(1, jsonStr1), Foo(2, jsonStr2), Foo(3, jsonStr3)).toDF

val schema = StructType(Seq(
  StructField("bar", StructType(Seq(
    StructField("foo", StringType, true)
    )), true),
  StructField("cars", ArrayType(StringType, true), true),
  StructField("daniel", StringType, true),
  StructField("fruits", ArrayType(StructType(Seq(
    StructField("a", StringType, true),
    StructField("dt", StringType, true),
    StructField("fruid", StringType, true),
    StructField("fruit", StringType, true)
  )), true), true)
))

df.
  withColumn("json_col", from_json($"json", schema)).
  select($"id", $"json_col.*").
  show
// +---+--------------------+------------------+------------------+--------------------+
// | id|                 bar|              cars|            daniel|              fruits|
// +---+--------------------+------------------+------------------+--------------------+
// |  1|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[null,null,null,...|
// |  2|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[null,banana,nul...|
// |  3|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,null,nul...|
// +---+--------------------+------------------+------------------+--------------------+

answered Mar 9, 2019 at 7:38

Leo C

22.5k3 gold badges28 silver badges42 bronze badges

3 Comments

Mantovani Over a year ago

I already knew this solution ,the JSON files are insanely huge, would be insane create the schema manually. I could use from_json if I Spark infer the schema automatic as it does with "spark.read.json". I was thinking in call spadk.read.json infer the schema and give as parameter to from_json. But I'm not sure it will be straightforward as sounds like, besides the override of the serialization.

Mantovani Over a year ago

ds.select($"id",from_json($"json",jsonDF.schema).alias("_json_path")).show

Leo C Over a year ago

@Mantovani, you could certainly get the schema from your jsonDF, which itself would require additional transformations to generate. For a large dataset with complex JSON schema, perhaps it would be best to create a JSON file with one single row of the JSON data, perform a spark.read.json and apply .schema for its schema.

Collectives™ on Stack Overflow

Apache Spark Read JSON With Extra Columns

2 Answers 2

Comments

3 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

3 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related