Read a json file in spark with garbage characters in the beginning

Question

I have a file that has data like this:

<1>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value11", "field21": "value12"} }
<2>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value21", "field2": "value22"} }
<3>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value31", "field2": "value32"} }

Normally in spark, I can just do spark.read.json("inputs.json"), but because of the garbage in the front of each row, I can't. Is there a way around this where I can chop off the front, or better yet--include the garbage as columns in my DataFrame?

Answer 1

Another method, where you get the schema dynamically by using a sample JSON record. The garbage string is parsed using the regular expression function regexp_extract()

Check this out:

scala> val df = Seq(( """<1>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value11", "field2": "value12"} }"""),
     | ("""<2>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value21", "field2": "value22"} }"""),
     | ("""<3>2019-03-20T20:59:59Z daily_report.txt[102852]: { "ts": "1553115599", "data": {"field1": "value31", "field2": "value32"} }""")).toDF("data_garb")
df: org.apache.spark.sql.DataFrame = [data_garb: string]

scala> val json_str = """{ "ts": "1553115599", "data": {"field1": "value11", "field2": "value12"} }"""
json_str: String = { "ts": "1553115599", "data": {"field1": "value11", "field2": "value12"} }

scala> val dfj = spark.read.json(Seq(json_str).toDS)
dfj: org.apache.spark.sql.DataFrame = [data: struct<field1: string, field2: string>, ts: string]

scala> dfj.schema
res44: org.apache.spark.sql.types.StructType = StructType(StructField(data,StructType(StructField(field1,StringType,true), StructField(field2,StringType,true)),true), StructField(ts,StringType,true))

scala> val df2=df.withColumn("newc",regexp_extract('data_garb,""".*?(\{.*)""",1)).withColumn("newc",from_json('newc,dfj.schema)).drop("data_garb")
df2: org.apache.spark.sql.DataFrame = [newc: struct<data: struct<field1: string, field2: string>, ts: string>]

scala> df2.show(false)
+--------------------------------+
|newc                            |
+--------------------------------+
|[[value11, value12], 1553115599]|
|[[value21, value22], 1553115599]|
|[[value31, value32], 1553115599]|
+--------------------------------+

The wildcard lets you select individual fields

scala>  df2.select($"newc.*").show(false)
+------------------+----------+
|data              |ts        |
+------------------+----------+
|[value11, value12]|1553115599|
|[value21, value22]|1553115599|
|[value31, value32]|1553115599|
+------------------+----------+


scala>

Or you can query the nested fields by explicitly mentioning them

scala> df2.select($"newc.ts",$"newc.data.field1",$"newc.data.field2").show(false)
+----------+-------+-------+
|ts        |field1 |field2 |
+----------+-------+-------+
|1553115599|value11|value12|
|1553115599|value21|value22|
|1553115599|value31|value32|
+----------+-------+-------+


scala>

Answer 2

You have to read the data as a Dataset[String] then parse the columns yourself. Once that is done, create a schema for your json data, and use sparks builtin from_json() function:

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

val ds = spark.createDataset(Seq(
    "<1>2019-03-20T20:59:59Z daily_report.txt[102852]: { \"ts\": \"1553115599\", \"data\": {\"field1\": \"value11\", \"field2\": \"value12\"} }",
    "<2>2019-03-20T20:59:59Z daily_report.txt[102852]: { \"ts\": \"1553115599\", \"data\": {\"field1\": \"value21\", \"field2\": \"value22\"} }",
    "<3>2019-03-20T20:59:59Z daily_report.txt[102852]: { \"ts\": \"1553115599\", \"data\": {\"field1\": \"value31\", \"field2\": \"value32\"} }"
))

//val ds = spark.read.text("inputs.txt").as[String]
val schema = StructType(List(StructField("ts", StringType), StructField("data", StructType(List(StructField("field1", StringType), StructField("field2", StringType))))))

val df = ds.map(r => {
    val j = r.indexOf("{")-1
    (r.substring(0, j), r.substring(j, r.length))
}).toDF("garbage", "json")

df.withColumn("data", from_json($"json", schema)).select("garbage", "data").show(false)

With your example data (field21 corrected to field2) you get:

+-------------------------------------------------+------------------------------+
|garbage                                          |data                          |
+-------------------------------------------------+------------------------------+
|<1>2019-03-20T20:59:59Z daily_report.txt[102852]:|[1553115599,[value11,value12]]|
|<2>2019-03-20T20:59:59Z daily_report.txt[102852]:|[1553115599,[value21,value22]]|
|<3>2019-03-20T20:59:59Z daily_report.txt[102852]:|[1553115599,[value31,value32]]|
+-------------------------------------------------+------------------------------+

With the schema:

root
 |-- garbage: string (nullable = true)
 |-- data: struct (nullable = true)
 |    |-- ts: string (nullable = true)
 |    |-- data: struct (nullable = true)
 |    |    |-- field1: string (nullable = true)
 |    |    |-- field2: string (nullable = true)

If you really don't need the garbage data, use the spark.read.json() you are already accustomed to by passing it a Dataset[String]. This doesn't require defining a schema, as it will be inferred:

val data = spark.read.json(df.select("json").as[String])