Spark Row zu JSON

Question

Ich möchte ein JSON aus einem Spark v.1.6 (mit Scala) Dataframe erstellen. Ich weiß, dass es die einfache Lösung gibt, df.toJSON zu tun.

Allerdings sieht mein Problem ein bisschen anders aus. Betrachten wir zum Beispiel einen Datenrahmen mit den folgenden Spalten:

| A |  B  | C1 |  C2 | C3 | 
------------------------------------------- 
| 1 | test  | ab | 22 | TRUE | 
| 2 | mytest | gh | 17 | FALSE | 

ich am Ende mit einem Datenrahmen haben möchte

| A |  B  |      C     | 
---------------------------------------------------------------- 
| 1 | test  | { "c1" : "ab", "c2" : 22, "c3" : TRUE } | 
| 2 | mytest | { "c1" : "gh", "c2" : 17, "c3" : FALSE } | 

wobei C eine JSON enthält C1, C2, C3. Leider weiß ich zur Kompilierzeit nicht wie der Datenframe aussieht (außer die Spalten A und B die immer "fixed" sind).

Aus dem Grund, warum ich das brauche: Ich verwende Protobuf zum Senden der Ergebnisse. Leider hat mein Datenframe manchmal mehr Spalten als erwartet und ich würde diese immer noch über Protobuf senden, aber ich möchte nicht alle Spalten in der Definition angeben.

Wie kann ich das erreichen?

Answer 1

Spark-2.1 ​​native Unterstützung für diesen Anwendungsfall (siehe #15354) haben sollte.

import org.apache.spark.sql.functions.to_json 
df.select(to_json(struct($"c1", $"c2", $"c3"))) 

Answer 2

Zuerst lässt C konvertieren ist ein struct:

val dfStruct = df.select($"A", $"B", struct($"C1", $"C2", $"C3").alias("C")) 

Dies kann Struktur JSONL umgewandelt werden unter Verwendung von toJSON wie zuvor:

dfStruct.toJSON.collect 
// Array[String] = Array(
// {"A":1,"B":"test","C":{"C1":"ab","C2":22,"C3":true}}, 
// {"A":2,"B":"mytest","C":{"C1":"gh","C2":17,"C3":false}}) 

Ich bin nicht bekannt, dass integrierte Methode, dass kann eine einzelne Spalte konvertieren, aber Sie können es entweder einzeln konvertieren und join oder verwenden Sie Ihre Lieblings-JSON-Parser in einer UDF.

case class C(C1: String, C2: Int, C3: Boolean) 

object CJsonizer { 
    import org.json4s._ 
    import org.json4s.JsonDSL._ 
    import org.json4s.jackson.Serialization 
    import org.json4s.jackson.Serialization.write 

    implicit val formats = Serialization.formats(org.json4s.NoTypeHints) 

    def toJSON(c1: String, c2: Int, c3: Boolean) = write(C(c1, c2, c3)) 
} 


val cToJSON = udf((c1: String, c2: Int, c3: Boolean) => 
    CJsonizer.toJSON(c1, c2, c3)) 

df.withColumn("c_json", cToJSON($"C1", $"C2", $"C3")) 

Answer 3

hier kein JSON-Parser, und sie paßt sich an Ihr Schema:

import org.apache.spark.sql.functions.{col, concat, concat_ws, lit} 

df.select(
    col(df.columns(0)), 
    col(df.columns(1)), 
    concat(
    lit("{"), 
    concat_ws(",",df.dtypes.slice(2, df.dtypes.length).map(dt => { 
     val c = dt._1; 
     val t = dt._2; 
     concat(
     lit("\"" + c + "\":" + (if (t == "StringType") "\""; else "") ), 
     col(c), 
     lit(if(t=="StringType") "\""; else "") 
    ) 
    }):_*), 
    lit("}") 
) as "C" 
).collect()