Funke: Programmgesteuert erstellen dataframe schema in scala

Habe ich ein kleineres dataset, das das Ergebnis einer Spark-job. Ich bin denken über das konvertieren in diesem dataset, um ein dataframe für die Bequemlichkeit am Ende der Arbeit, aber haben gekämpft, um korrekt definieren Sie das schema. Das problem ist das Letzte Feld unten (topValues); es ist ein ArrayBuffer von Tupeln -- Tasten und zählt.

  val innerSchema =
    StructType(
      Array(
        StructField("value", StringType),
        StructField("count", LongType)
      )
    )
  val outputSchema =
    StructType(
      Array(
        StructField("name", StringType, nullable=false),
        StructField("index", IntegerType, nullable=false),
        StructField("count", LongType, nullable=false),
        StructField("empties", LongType, nullable=false),
        StructField("nulls", LongType, nullable=false),
        StructField("uniqueValues", LongType, nullable=false),
        StructField("mean", DoubleType),
        StructField("min", DoubleType),
        StructField("max", DoubleType),
        StructField("topValues", innerSchema)
      )
    )

  val result = stats.columnStats.map{ c =>
    Row(c._2.name, c._1, c._2.count, c._2.empties, c._2.nulls, c._2.uniqueValues, c._2.mean, c._2.min, c._2.max, c._2.topValues.topN)
  }

  val rdd = sc.parallelize(result.toSeq)

  val outputDf = sqlContext.createDataFrame(rdd, outputSchema)

  outputDf.show()

Die Fehler, die ich bekommen hab ist ein MatchError: scala.MatchError: ArrayBuffer((10,2), (20,3), (8,1)) (of class scala.collection.mutable.ArrayBuffer)

Wenn ich Debuggen und inspizieren meine Objekte, ich sehe dies:

rdd: ParallelCollectionRDD[2]
rdd.data: "ArrayBuffer" size = 2
rdd.data(0): [age,2,6,0,0,3,14.666666666666666,8.0,20.0,ArrayBuffer((10,2), (20,3), (8,1))]
rdd.data(1): [gender,3,6,0,0,2,0.0,0.0,0.0,ArrayBuffer((M,4), (F,2))]

Es scheint mir, ich habe genau beschrieben, die ArrayBuffer von Tupeln in meinem innerSchema, aber Funke nicht einverstanden ist.

Irgendeine Idee, wie ich das definieren das schema?

Es wäre nützlich, wenn Sie ein Beispiel geben, Daten oder zumindest eine genaue Art der rdd.

InformationsquelleAutor Stuart | 2016-03-30

val rdd = sc.parallelize(Array(Row(ArrayBuffer(1,2,3,4))))
val df = sqlContext.createDataFrame(
  rdd,
  StructType(Seq(StructField("arr", ArrayType(IntegerType, false), false)
)

df.printSchema
root
 |-- arr: array (nullable = false)
 |    |-- element: integer (containsNull = false)

df.show
+------------+
|         arr|
+------------+
|[1, 2, 3, 4]|
+------------+

Ja, ArrayType ist der richtige Ansatz. Danke! Meine Letzte schema ist in meiner Antwort.

InformationsquelleAutor David Griffin

Als David wies darauf hin, ich brauchte, um ein ArrayType. Funke ist zufrieden mit diesem:

  val outputSchema =
    StructType(
      Array(
        StructField("name", StringType, nullable=false),
        StructField("index", IntegerType, nullable=false),
        StructField("count", LongType, nullable=false),
        StructField("empties", LongType, nullable=false),
        StructField("nulls", LongType, nullable=false),
        StructField("uniqueValues", LongType, nullable=false),
        StructField("mean", DoubleType),
        StructField("min", DoubleType),
        StructField("max", DoubleType),
        StructField("topValues", ArrayType(StructType(Array(
          StructField("value", StringType),
          StructField("count", LongType)
        ))))
      )
    )

InformationsquelleAutor Stuart

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


val searchPath = "/path/to/.csv"
val columns = "col1,col2,col3,col4,col5,col6,col7"
val fields = columns.split(",").map(fieldName => StructField(fieldName, StringType, 
nullable = true))
val customSchema = StructType(fields)
var dfPivot =spark.read.format("com.databricks.spark.csv").option("header","false").option("inferSchema", "false").schema(customSchema).load(searchPath)

Beim laden der Daten mit benutzerdefinierten schema wird viel schneller im Vergleich zum laden von Daten mit Standard-schema

InformationsquelleAutor Arun Goudar

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.