in einem Datenrahmen in Pyspark

Question

Ich habe einen Datenrahmen mit einigen Attributen und es hat den nächsten Auftritt:

+-------+-------+ 
| Atr1 | Atr2 | 
+-------+-------+ 
| 3,06 | 4,08 | 
| 3,03 | 4,08 | 
| 3,06 | 4,08 | 
| 3,06 | 4,08 | 
| 3,06 | 4,08 | 
| ... | ... | 
+-------+-------+ 

Wie Sie sehen können, werden die Werte des ATR1 und ATR2 der Datenrahmen sind Zahlen das hat ein ',' Zeichen. Dies liegt daran, dass ich diese Daten aus einer CSV-Datei geladen habe, in der die Dezimalstellen der DoubleType-Zahlen durch "," dargestellt wurden.

Wenn ich die Daten in einen Datenrahmen laden die Werte in String gegossen werden, so bewarb ich mich ein Casting von String für diese Attribute so DoubleType:

df = df.withColumn("Atr1", df["Atr1"].cast(DoubleType())) 
df = df.withColumn("Atr2", df["Atr2"].cast(DoubleType())) 

Aber wenn ich es tun, sind die Werte auf null umgewandelt

+-------+-------+ 
| Atr1 | Atr2 | 
+-------+-------+ 
| null | null | 
| null | null | 
| null | null | 
| null | null | 
| null | null | 
| ... | ... | 
+-------+-------+ 

ich denke, dass der Grund dafür ist, dass DoubleType Dezimalstellen müssen getrennt werden ‚‘ statt durch ','. Aber ich habe keine Möglichkeit, die CSV-Datei zu bearbeiten, deshalb möchte ich die ',' Zeichen aus dem Dataframe durch '.' und wenden Sie dann das Casting auf DoubleType an.

Wie könnte ich es tun?

Answer 1

Sie können dieses Problem einfach lösen, indem Sie eine benutzerdefinierte Funktion verwenden.

from pyspark.sql.functions import UserDefinedFunction 
from pyspark.sql.functions import * 

data = [Row(Atr1="3,06", Atr2="4,08"), 
     Row(Atr1="3,06", Atr2="4,08"), 
     Row(Atr1="3,06", Atr2="4,08")] 

df = sqlContext.createDataFrame(data) 

# Create an user defined function to replace ',' for '.' 
udf = UserDefinedFunction(lambda x: x.replace(",","."), StringType()) 

out = df 
    .withColumn("Atr1", udf(col("Atr1")).cast(DoubleType())) 
    .withColumn("Atr2", udf(col("Atr2")).cast(DoubleType())) 

############################################################## 
out.show() 

+----+----+ 
|Atr1|Atr2| 
+----+----+ 
|3.06|4.08| 
|3.06|4.08| 
|3.06|4.08| 
+----+----+ 

############################################################## 

out.printSchema() 

root 
|-- Atr1: double (nullable = true) 
|-- Atr2: double (nullable = true) 

EDIT: Kompaktere Lösung von Kommentaren folgenden Vorschlag.

from pyspark.sql.functions import UserDefinedFunction 
from pyspark.sql.functions import * 

udf = UserDefinedFunction(lambda x: float(x.replace(",",".")), DoubleType()) 

out = df 
    .withColumn("Atr1", udf(col("Atr1"))) 
    .withColumn("Atr2", udf(col("Atr2"))) 

Answer 2

Sie können es auch mit nur SQL tun.

val df = sc.parallelize(Array(
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08"), 
     ("3,06", "4,08") 
    )).toDF("a", "b") 

df.registerTempTable("test") 

val doubleDF = sqlContext.sql("select cast(trim(regexp_replace(a , ',' , '.')) as double) as a from test ") 

doubleDF.show 
+----+ 
| a| 
+----+ 
|3.06| 
|3.06| 
|3.06| 
|3.06| 
|3.06| 
|3.06| 
|3.06| 
|3.06| 
+----+ 

doubleDF.printSchema 
root 
|-- a: double (nullable = true) 

Answer 3

Nehmen wir an, Sie haben:

sdf.show() 
+-------+-------+ 
| Atr1| Atr2| 
+-------+-------+ 
| 3,06 | 4,08 | 
| 3,03 | 4,08 | 
| 3,06 | 4,08 | 
| 3,06 | 4,08 | 
| 3,06 | 4,08 | 
+-------+-------+ 

Dann wird der folgende Code wird das gewünschte Ergebnis:

strToDouble = udf(lambda x: float(x.replace(",",".")), DoubleType()) 

sdf = sdf.withColumn("Atr1", strToDouble(sdf['Atr1'])) 
sdf = sdf.withColumn("Atr2", strToDouble(sdf['Atr2'])) 

sdf.show() 
+----+----+ 
|Atr1|Atr2| 
+----+----+ 
|3.06|4.08| 
|3.03|4.08| 
|3.06|4.08| 
|3.06|4.08| 
|3.06|4.08| 
+----+----+ 

Answer 4

ist es möglich, den Spaltennamen als Parameter an die Spalte übergeben() Funktion in Ihrem Beispielcode? Etwas wie dieses:

# Create an user defined function to replace ',' for '.' 
udf = UserDefinedFunction(lambda x: x.replace(",","."), StringType()) 

col_name1 = "Atr1" 
col_name2 = "Atr2" 

out = df 
    .withColumn(col_name1, udf(col(col_name1)).cast(DoubleType())) 
    .withColumn(col_name2, udf(col(col_name2)).cast(DoubleType()))