Wie behandelt man die Null/Leer-Werte auf einem Datenrahmen Spark/Scala

Question

Ich habe eine CSV-Datei und ich verarbeite seine Daten.

Ich arbeite mit Datenrahmen, und ich berechne Mittelwert, Minimum, Maximum, Mittelwert, Summe jeder Spalte basierend auf einigen Bedingungen. Die Daten jeder Spalte können leer oder null sein.

Ich habe festgestellt, dass in einigen Fällen habe ich als max, oder summe einen Nullwert anstelle einer Zahl. Oder ich habe in max() eine Zahl, die weniger ist als die Ausgabe, die das min() zurückgibt.

Ich möchte nicht die Null/leere Werte durch andere ersetzen. Das einzige, was ich getan habe, ist diese 2 Optionen in CSV zu verwenden:

.option("nullValue", "null") 
.option("treatEmptyValuesAsNulls", "true") 

Gibt es eine Möglichkeit, dieses Problem zu umgehen? Sind alle schon einmal mit diesem Problem konfrontiert worden? Ist es ein Problem von Datentypen?

Ich laufe etwas wie folgt aus:

data.agg(mean("col_name"), stddev("col_name"),count("col_name"), 
     min("col_name"), max("col_name")) 

Ansonsten kann ich der Ansicht, dass es sich um ein Problem in meinem Code.

Answer 1

Im Gegensatz zu einem der Kommentare ist es nicht wahr, dass Nullen ignoriert werden. Hier ist ein Ansatz:

max(coalesce(col_name,Integer.MinValue)) 
min(coalesce(col_name,Integer.MaxValue)) 

Das wird noch ein Problem haben, wenn es waren nur Nullwerte: Sie müssen Min/MaxValue konvertieren auf null oder was auch immer Sie verwenden möchten, um „keine gültige/Nicht -null Einträge ".

Answer 2

Ich habe einige Untersuchungen zu dieser Frage durchgeführt, und das Ergebnis zeigt, dass die Funktionen mean, max, min Nullwerte ignorieren. Unten finden Sie den Testcode und die Ergebnisse. Umwelt: Scala, Funken 1.6.1 Hadoop 2.6.0

import org.apache.spark.sql.{Row} 
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType} 
import org.apache.spark.sql.types._ 
import org.apache.spark.{SparkConf, SparkContext} 

val row1 =Row("1", 2.4, "2016-12-21") 
val row2 = Row("1", None, "2016-12-22") 
val row3 = Row("2", None, "2016-12-23") 
val row4 = Row("2", None, "2016-12-23") 
val row5 = Row("3", 3.0, "2016-12-22") 
val row6 = Row("3", 2.0, "2016-12-22") 
val theRdd = sc.makeRDD(Array(row1, row2, row3, row4, row5, row6)) 

val schema = StructType(StructField("key", StringType, false) :: 
        StructField("value", DoubleType, true) :: 
        StructField("d", StringType, false) :: Nil) 
val df = sqlContext.createDataFrame(theRdd, schema) 

df.show() 

df.agg(mean($"value"), max($"value"), min($"value")).show() 

df.groupBy("key").agg(mean($"value"), max($"value"), min($"value")).show() 


Output: 
+---+-----+----------+ 
|key|value|   d| 
+---+-----+----------+ 
| 1| 2.4|2016-12-21| 
| 1| null|2016-12-22| 
| 2| null|2016-12-23| 
| 2| null|2016-12-23| 
| 3| 3.0|2016-12-22| 
| 3| 2.0|2016-12-22| 
+---+-----+----------+ 
+-----------------+----------+----------+ 
|  avg(value)|max(value)|min(value)| 
+-----------------+----------+----------+ 
|2.466666666666667|  3.0|  2.0| 
+-----------------+----------+----------+ 
+---+----------+----------+----------+ 
|key|avg(value)|max(value)|min(value)| 
+---+----------+----------+----------+ 
| 1|  2.4|  2.4|  2.4| 
| 2|  null|  null|  null| 
| 3|  2.5|  3.0|  2.0| 
+---+----------+----------+----------+ 

Vom Ausgang Sie, dass der Mittelwert sehen, max, min Funktionen auf Spalte 'Wert' der Gruppenschlüssel = '1' gibt '2.4' anstelle von null, was zeigt, dass die Nullwerte in diesen Funktionen ignoriert wurden. Wenn die Spalte jedoch nur Nullwerte enthält, geben diese Funktionen Nullwerte zurück.