Spark Zeile JSON

Ich möchte eine JSON von einer Spark-v. 1.6 (mit scala) dataframe. Ich weiß, dass es die einfache Lösung zu tun df.toJSON.

Allerdings ist mein problem etwas anders aussieht. Betrachten Sie zum Beispiel einen dataframe mit den folgenden Spalten:

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

Ich würde gerne am Ende ein dataframe mit 

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

wobei C ein JSON mit C1, C2, C3. Leider habe ich zur compile-Zeit, weiß ich nicht, was das dataframe aussieht (mit Ausnahme der Spalten A und B sind immer "Feste").

Als für der Grund warum brauche ich das: ich bin mit Protobuf für das senden um die Ergebnisse. Leider ist mein dataframe manchmal hat mehr Spalten als erwartet und ich würde immer noch senden Sie diese per Protobuf, aber ich will nicht angeben, werden alle Spalten in der definition.

Wie kann ich das erreichen?

  • noch ein dataframe
  • Nein, sorry, ich meine eher, wie man hinzufügen C1, C2, C3 als JSON-string-Spalte, um die bestehenden dataframe. Ich habe aktualisiert die post zu klären, für die version von Spark und scala als Sprache.
  • Sorry! Sicher, ich gerade aktualisiert die Frage (zusammen mit ein Grund, warum ich möchte, um dies zu erreichen) und fügte ein Beispiel.
InformationsquelleAutor navige | 2016-03-22
  1. 18

    Funke 2.1 sollte die native Unterstützung für diesen Anwendungsfall (siehe #15354).

    import org.apache.spark.sql.functions.to_json
df.select(to_json(struct($"c1", $"c2", $"c3")))
    InformationsquelleAutor Michael Armbrust
  2. 5

    Erste ermöglicht das konvertieren von C ist, um eine struct:

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

    Diese Struktur umgewandelt werden können, um JSONL mit toJSON wie vor:

    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 kenne keine integrierte Methode, die die konvertieren kann eine einzelne Spalte, sondern Sie können entweder konvertieren Sie individuell und join oder verwenden Sie Ihre Lieblings-JSON-parser in eine 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"))
    • Tatsächlich, meine Frage ist wirklich über den zweiten Teil wie konvertiert man die einzelnen Spalten auf JSON. Sie erwähnen join-ing Spalten, aber nicht wirklich funktionieren, wie ich auf der einen Seite eine RDD[String] und auf der anderen Seite eine DataFrame
    • Wie er sagt, nur mit einem UDF. Sie haben nicht einmal zu verwenden, eine ausgewachsene JSON-parser in der UDF - Sie können einfach Handwerk eine JSON-string on the fly mit map und mkString. Sie werden wahrscheinlich brauchen, um zu verwenden DataFrame.columns oder möglicherweise DataFrame.dtypes sowohl Handwerk der select Erklärung und als Grundlage der map im UDF.
    • Ich Stimme mit @DavidGriffin - udf werden kann, die einfachste Lösung hier. Und Jackson und json4s sind bereits gezogen, mit anderen Abhängigkeiten.
    • Mein problem mit dem JSON-Parser, die ich gesehen habe ist, dass Sie brauchen, um im Voraus wissen, wie das schema aussieht -- wie mit deiner Lösung @zero323 -- es funktioniert nur das für die jeweiligen Spalten. Was ist, wenn die Namen anders wären? Was, wenn es mehr als 3 Spalten?
    • Das einzige problem das ich sehe ist, dass Row ist extrem hässlich Datenstruktur. Ansonsten können Sie erstellen Sie einfach eine beliebige komplexe AST mit Lift / json4s und konvertieren von JSON. Aber ehrlich gesagt, es ist zu viel Aufwand um es in ein SO Antwort.
    • Row hässlich ist aus dem gleichen Grund ich hasse den Umgang mit JSON in der Scala-es ist ein Kampf der Kulturen Loosey, goosey vs starke, statische Typisierung. SQL ist loosey goosey-Sie sind ein select Weg von der Definition der neuen Art -- damit Row ist chaotisch. Avro ist GenericRecord hat das gleiche problem.

    InformationsquelleAutor
  3. 5

    Hier keine JSON-parser, und es passt zu deinem 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()
    • sieht ein bisschen hacky, aber es funktioniert 🙂
    • Yup und yup. JSON ist aber hacky im Allgemeinen, wenn Sie mich Fragen.
    InformationsquelleAutor David Griffin
  4. 1

    Ich mit diesem Befehl lösen Sie die to_json problem:

    output_df = (df.select(to_json(struct(col("*"))).alias("content")))
    InformationsquelleAutor Cyanny
Schreibe einen Kommentar