Прочитайте файл JSON в спарк с мусорных символов в начале

Question

У меня есть файл с такими данными:

<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"} }

Обычно в свечах я могу просто сделать spark.read.json("inputs.json"), но из-за мусора в начале каждого ряда я не могу. Есть ли способ обойти это, где я могу отрубить фронт, или, что еще лучше, включить мусор в виде столбцов в моем DataFrame?

Answer 1

Вы должны прочитать данные как Dataset[String], а затем проанализировать столбцы самостоятельно. Как только это будет сделано, создайте схему для ваших данных json и используйте встроенную функцию sparks from_json():

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)

С вашими примерами (field21 скорректирован на field2) вы получите:

+-------------------------------------------------+------------------------------+
|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]]|
+-------------------------------------------------+------------------------------+

со схемой:

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)

Если вам действительно не нужны данные garbage, используйте spark.read.json(), к которому вы уже привыкли, передав ему Dataset[String]. Это не требует определения схемы, так как это будет выведено:

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

Answer 2

Другой метод, где вы получаете схему динамически, используя пример записи JSON.Строка мусора анализируется с помощью функции регулярного выражения regexp_extract ()

Проверьте это:

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]|
+--------------------------------+

Подстановочный знак позволяет выбирать отдельные поля

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


scala>

Или выможет запрашивать вложенные поля, явно упоминая их

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>