Zugriff auf JSON-Schema-Informationen in SparkDataset Api bei Verwendung von Plain Map/Reduce-Funktionen?

Question

Da gibt es einen Datensatz von Nachrichten, die von folgendem Code definiert:

case class Message(id: Int, value: String) 

    var messages = Seq(
    (0, """{"action":"update","timestamp":"2017-10-05T23:01:19Z"}"""), 
    (1, """{"action":"update","timestamp":"2017-10-05T23:01:19Z"}""") 
).toDF("id", "value").as[Message] 


var schema = new StructType().add("action", StringType).add("timestamp", TimestampType) 

var res = messages.select(
    from_json(col("value").cast("string"), schema) 
) 

+------------------------------------+ 
|jsontostructs(CAST(value AS STRING))| 
+------------------------------------+ 
|    [update,2017-10-0...| 
|    [update,2017-10-0...| 

Was ist der beste Weg, um die Schema-Informationen in einer einfachen Map-Funktion zugreifen. Die Funktion selbst gibt eine Zeile zurück, die alle Type-Infos verloren hat. Um die Werte zu erreichen muss man den Typen angeben, wieder z

res.head().getStruct(0).getValuesMap[TimestampType](Seq("timestamp")) 
=> Map[String,org.apache.spark.sql.types.TimestampType] = Map(timestamp -> 2017-10-06 01:01:19.0) 

oder

res.head().getStruct(0).getString(0) 
=> res20: String = update 

Gibt es einen besseren Weg, um die rohen json Daten ohne Funken SQL-Aggregationsfunktionen zugreifen?

Answer 1

Als Faustregel:

• Zur Sammlung API (map, flatMap, mapPartitions, groupByKey, etc.) verwenden stark API getippt verwenden - definieren Satzart (Fall-Klasse am besten funktioniert), die reflektiert das Schema und verwendet Encoders Dinge zu konvertieren hin und her:

  case class Value(action: String, timestamp: java.sql.Timestamp) 
  case class ParsedMessage(id: Int, value: Option[Value]) 
  
  messages.select(
      $"id", from_json(col("value").cast("string"), schema).alias("value") 
  ).as[ParsedMessage].map(???) 
  

• mit Dataset[Row] Aufenthalt mit hohem Niveau SQL/DataFrame AP I (select, where, agg, groupBy)