Wie kann ich Spaltentypen im DataFrame von Spark SQL ändern?

151

Angenommen, ich mache so etwas wie:

val df = sqlContext.load("com.databricks.spark.csv", Map("path" -> "cars.csv", "header" -> "true"))
df.printSchema()

root
 |-- year: string (nullable = true)
 |-- make: string (nullable = true)
 |-- model: string (nullable = true)
 |-- comment: string (nullable = true)
 |-- blank: string (nullable = true)

df.show()
year make  model comment              blank
2012 Tesla S     No comment
1997 Ford  E350  Go get one now th...

Aber ich wollte wirklich das yearas Int(und vielleicht einige andere Spalten transformieren).

Das Beste, was ich mir einfallen lassen konnte, war

df.withColumn("year2", 'year.cast("Int")).select('year2 as 'year, 'make, 'model, 'comment, 'blank)
org.apache.spark.sql.DataFrame = [year: int, make: string, model: string, comment: string, blank: string]

das ist ein bisschen verworren.

Ich komme aus R und bin es gewohnt, schreiben zu können, z

df2 <- df %>%
   mutate(year = year %>% as.integer,
          make = make %>% toupper)

Ich vermisse wahrscheinlich etwas, da es in Spark / Scala einen besseren Weg geben sollte, dies zu tun ...

kevinykuo
quelle
Ich mag diese Art und Weise spark.sql ("SELECT STRING (NULLIF (Spalte, '')) als column_string")
Eric Bellet

Antworten:

141

Bearbeiten: Neueste Version

Seit Spark 2.x können Sie verwenden .withColumn. Überprüfen Sie die Dokumente hier:

https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Dataset@withColumn(colName:String,col:org.apache.spark.sql.Column) : org.apache.spark.sql.DataFrame

Älteste Antwort

Seit Spark Version 1.4 können Sie die Cast-Methode mit DataType auf die Spalte anwenden:

import org.apache.spark.sql.types.IntegerType
val df2 = df.withColumn("yearTmp", df.year.cast(IntegerType))
    .drop("year")
    .withColumnRenamed("yearTmp", "year")

Wenn Sie SQL-Ausdrücke verwenden, können Sie auch Folgendes tun:

val df2 = df.selectExpr("cast(year as int) year", 
                        "make", 
                        "model", 
                        "comment", 
                        "blank")

Weitere Informationen finden Sie in den Dokumenten: http://spark.apache.org/docs/1.6.0/api/scala/#org.apache.spark.sql.DataFrame

msemelman
quelle
4
Warum haben Sie withColumn gefolgt von drop verwendet? Ist es nicht einfacher, einfach withColumn mit dem ursprünglichen Spaltennamen zu verwenden?
Ameba Spugnosa
@AmebaSpugnosa Ich denke, als ich es benutzte, stürzte Spark ab, wenn es wiederholte Spaltennamen hatte. Nicht wenn Sie sie erstellen, sondern wenn Sie sie verwenden.
Msemelman
5
Es ist nicht erforderlich, eine Spalte zu löschen, gefolgt von einer Umbenennung. Sie können in einer Zeile tundf.withColumn("ctr", temp("ctr").cast(DecimalType(decimalPrecision, decimalScale)))
ruhong
1
Wird in diesem Fall eine vollständig neue Datenrahmenkopie erstellt, um eine Spalte neu zu erstellen? Vermisse ich etwas Oder gibt es vielleicht eine Optimierung hinter den Kulissen?
user1814008
5
Wenn Sie sich an die Dokumente von halten Spark 2.x, df.withColumn(..)können Sie je nach colNameArgument eine Spalte hinzufügen oder ersetzen
y2k-shubham
89

[EDIT: März 2016: Danke für die Stimmen! Obwohl wirklich, das ist nicht die beste Antwort, ich glaube , die Lösungen auf Basis von withColumn, withColumnRenamedund castlegte durch msemelman, Martin Senne und andere sind einfacher und sauberer].

Ich denke, Ihr Ansatz ist in Ordnung. Denken Sie daran, dass ein Spark DataFrameeine (unveränderliche) RDD von Zeilen ist. Wir ersetzen also nie wirklich eine Spalte, sondern erstellen DataFramejedes Mal eine neue durch ein neues Schema.

Angenommen, Sie haben eine Original-DF mit dem folgenden Schema:

scala> df.printSchema
root
 |-- Year: string (nullable = true)
 |-- Month: string (nullable = true)
 |-- DayofMonth: string (nullable = true)
 |-- DayOfWeek: string (nullable = true)
 |-- DepDelay: string (nullable = true)
 |-- Distance: string (nullable = true)
 |-- CRSDepTime: string (nullable = true)

Und einige UDFs, die in einer oder mehreren Spalten definiert sind:

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

val toInt    = udf[Int, String]( _.toInt)
val toDouble = udf[Double, String]( _.toDouble)
val toHour   = udf((t: String) => "%04d".format(t.toInt).take(2).toInt ) 
val days_since_nearest_holidays = udf( 
  (year:String, month:String, dayOfMonth:String) => year.toInt + 27 + month.toInt-12
 )

Das Ändern von Spaltentypen oder sogar das Erstellen eines neuen DataFrame aus einem anderen kann folgendermaßen geschrieben werden:

val featureDf = df
.withColumn("departureDelay", toDouble(df("DepDelay")))
.withColumn("departureHour",  toHour(df("CRSDepTime")))
.withColumn("dayOfWeek",      toInt(df("DayOfWeek")))              
.withColumn("dayOfMonth",     toInt(df("DayofMonth")))              
.withColumn("month",          toInt(df("Month")))              
.withColumn("distance",       toDouble(df("Distance")))              
.withColumn("nearestHoliday", days_since_nearest_holidays(
              df("Year"), df("Month"), df("DayofMonth"))
            )              
.select("departureDelay", "departureHour", "dayOfWeek", "dayOfMonth", 
        "month", "distance", "nearestHoliday")            

was ergibt:

scala> df.printSchema
root
 |-- departureDelay: double (nullable = true)
 |-- departureHour: integer (nullable = true)
 |-- dayOfWeek: integer (nullable = true)
 |-- dayOfMonth: integer (nullable = true)
 |-- month: integer (nullable = true)
 |-- distance: double (nullable = true)
 |-- nearestHoliday: integer (nullable = true)

Dies kommt Ihrer eigenen Lösung ziemlich nahe. Wenn Sie die Typänderungen und andere Transformationen einfach getrennt halten udf val, wird der Code einfach lesbarer und wiederverwendbarer.

Svend
quelle
26
Dies ist weder sicher noch effizient. Nicht sicher, da ein einzelner NULLoder fehlerhafter Eintrag einen ganzen Job zum Absturz bringt. Nicht effizient, da UDFs für Catalyst nicht transparent sind. Die Verwendung von UDFs für komplexe Operationen ist in Ordnung, aber es gibt keinen Grund, diese für das Gießen von Grundtypen zu verwenden. Deshalb haben wir castMethode (siehe eine Antwort von Martin Senne ). Die Transparenz für Catalyst erfordert mehr Arbeit, aber die grundlegende Sicherheit ist nur eine Frage des Puttens Tryund OptionArbeitens.
Null323
Ich habe nichts im Zusammenhang mit der Konvertierung von Zeichenfolgen bis heute gesehen, zum Beispiel "05-APR-2015"
dbspace
3
Gibt es eine Möglichkeit, Ihren withColumn()Abschnitt auf einen generischen zu reduzieren , der alle Spalten durchläuft?
Boern
Dank zero323 habe ich beim Lesen herausgefunden, warum die udf-Lösung hier abstürzt. Einige Kommentare sind besser als einige Antworten auf SO :)
Simon Dirmeier
Gibt es eine Möglichkeit, die beschädigte Zeile kennenzulernen, dh Datensätze, die während des Castings Spalten mit falschen Datentypen aufweisen? Als Cast-Funktion werden diese Felder als null gesetzt
Etisha
65

Da die castOperation für Spark verfügbar ist Column(und ich persönlich die udfvon @ Svendan dieser Stelle vorgeschlagenen nicht bevorzuge ), wie wäre es mit:

df.select( df("year").cast(IntegerType).as("year"), ... )

auf den gewünschten Typ umwandeln? Als ordentliche Nebenwirkung, Werte nicht gießbaren / „konvertierbar“ in diesem Sinne wird worden null.

Wenn Sie dies als Hilfsmethode benötigen , verwenden Sie:

object DFHelper{
  def castColumnTo( df: DataFrame, cn: String, tpe: DataType ) : DataFrame = {
    df.withColumn( cn, df(cn).cast(tpe) )
  }
}

welches verwendet wird wie:

import DFHelper._
val df2 = castColumnTo( df, "year", IntegerType )
Martin Senne
quelle
2
Können Sie mir raten, wie ich vorgehen soll, wenn ich eine ganze Reihe von Spalten umwandeln und umbenennen muss (ich habe 50 Spalten und bin relativ neu in Scala, nicht sicher, wie ich am besten vorgehen soll, ohne eine massive Duplizierung zu erstellen)? Einige Spalten sollten String bleiben, andere sollten in Float umgewandelt werden.
Dmitry Smirnov
wie man einen String in ein Datum konvertiert, zum Beispiel "25-APR-2016" in der Spalte und "20160302"
dbspace
@DmitrySmirnov Hast du jemals eine Antwort bekommen? Ich habe die gleiche Frage. ;)
Evan Zamir
@EvanZamir leider nicht, ich habe eine Menge Operationen durchgeführt, um Daten in anderen Schritten als rdd verwenden zu können. Ich frage mich, ob dies heutzutage einfacher wurde :)
Dmitry Smirnov
59

Erstens , wenn Sie wollen , Typ umwandeln, dann das:

import org.apache.spark.sql
df.withColumn("year", $"year".cast(sql.types.IntegerType))

Bei gleichem Spaltennamen wird die Spalte durch eine neue ersetzt. Sie müssen keine Schritte hinzufügen und löschen.

Zweitens , etwa Scala vs R .
Dies ist der Code, der RI am ähnlichsten ist:

val df2 = df.select(
   df.columns.map {
     case year @ "year" => df(year).cast(IntegerType).as(year)
     case make @ "make" => functions.upper(df(make)).as(make)
     case other         => df(other)
   }: _*
)

Obwohl die Codelänge etwas länger ist als die von R. Das hat nichts mit der Ausführlichkeit der Sprache zu tun. In R mutateist dies eine spezielle Funktion für R-Datenrahmen, während Sie in Scala dank ihrer Ausdruckskraft problemlos eine Ad-hoc-Funktion verwenden können.
Mit anderen Worten, es werden bestimmte Lösungen vermieden, da das Sprachdesign gut genug ist, um schnell und einfach Ihre eigene Domain-Sprache zu erstellen.


Randnotiz: df.columnsist überraschenderweise ein Array[String]statt Array[Column], vielleicht wollen sie, dass es wie der Datenrahmen von Python Pandas aussieht.

WeiChing 林 煒 清
quelle
1
Könnten Sie bitte das Äquivalent für pyspark angeben?
Harit Vishwakarma
Ich erhalte "illegalen Start der Definition" .withColumn ("age", $ "age" .cast (sql.types.DoubleType)) für mein Feld "age". Irgendein Vorschlag?
BlueDolphin
Müssen Sie den Datenrahmen .cache (), wenn wir diese Konvertierungen aus Leistungsgründen für viele Spalten durchführen, oder ist dies nicht erforderlich, da Spark sie optimiert?
Skjagini
Der Import kann sein import org.apache.spark.sql.types._und dann statt sql.types.IntegerTypenur IntegerType.
nessa.gp
17

Sie können es verwenden selectExpr, um es ein wenig sauberer zu machen:

df.selectExpr("cast(year as int) as year", "upper(make) as make",
    "model", "comment", "blank")
dnlbrky
quelle
14

Java-Code zum Ändern des Datentyps des DataFrame von String in Integer

df.withColumn("col_name", df.col("col_name").cast(DataTypes.IntegerType))

Der vorhandene (String-Datentyp) wird einfach in Integer umgewandelt.

manishbelsare
quelle
1
Da ist kein DataTypesrein sql.types! es ist DataType. Darüber hinaus kann man einfach importieren IntegerTypeund gießen.
Ehsan M. Kermani
@ EhsanM.Kermani eigentlich DatyaTypes.IntegerType ist eine legitime Referenz.
Cupitor
1
@Cupitor war DataTypes.IntegerTypefrüher im DeveloperAPI-Modus und es ist stabil in v.2.1.0
Ehsan M. Kermani
Dies ist die beste Lösung!
Simon Dirmeier
8

Um das Jahr von Zeichenfolge in int zu konvertieren, können Sie dem CSV-Reader die folgende Option hinzufügen: "inferSchema" -> "true", siehe DataBricks-Dokumentation

Peter Rose
quelle
5
Dies funktioniert gut, aber der Haken ist, dass der Leser einen zweiten Durchgang Ihrer Datei machen muss
Beefyhalo
@beefyhalo absolut genau richtig, gibt es einen Weg, das zu umgehen?
Ayush
6

Dies funktioniert also nur dann wirklich, wenn Sie Probleme beim Speichern auf einem JDBC-Treiber wie sqlserver haben. Es ist jedoch sehr hilfreich bei Fehlern, die bei der Syntax und den Typen auftreten.

import org.apache.spark.sql.jdbc.{JdbcDialects, JdbcType, JdbcDialect}
import org.apache.spark.sql.jdbc.JdbcType
val SQLServerDialect = new JdbcDialect {
  override def canHandle(url: String): Boolean = url.startsWith("jdbc:jtds:sqlserver") || url.contains("sqlserver")

  override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
    case StringType => Some(JdbcType("VARCHAR(5000)", java.sql.Types.VARCHAR))
    case BooleanType => Some(JdbcType("BIT(1)", java.sql.Types.BIT))
    case IntegerType => Some(JdbcType("INTEGER", java.sql.Types.INTEGER))
    case LongType => Some(JdbcType("BIGINT", java.sql.Types.BIGINT))
    case DoubleType => Some(JdbcType("DOUBLE PRECISION", java.sql.Types.DOUBLE))
    case FloatType => Some(JdbcType("REAL", java.sql.Types.REAL))
    case ShortType => Some(JdbcType("INTEGER", java.sql.Types.INTEGER))
    case ByteType => Some(JdbcType("INTEGER", java.sql.Types.INTEGER))
    case BinaryType => Some(JdbcType("BINARY", java.sql.Types.BINARY))
    case TimestampType => Some(JdbcType("DATE", java.sql.Types.DATE))
    case DateType => Some(JdbcType("DATE", java.sql.Types.DATE))
    //      case DecimalType.Fixed(precision, scale) => Some(JdbcType("NUMBER(" + precision + "," + scale + ")", java.sql.Types.NUMERIC))
    case t: DecimalType => Some(JdbcType(s"DECIMAL(${t.precision},${t.scale})", java.sql.Types.DECIMAL))
    case _ => throw new IllegalArgumentException(s"Don't know how to save ${dt.json} to JDBC")
  }
}

JdbcDialects.registerDialect(SQLServerDialect)
Ben Jarman
quelle
Können Sie mir helfen, denselben Code in Java zu implementieren? und wie man den customJdbcDialect in DataFrame registriert
abhijitcaps
Schön, dass ich das gleiche mit Vertica gemacht habe, aber seit Spark 2.1. JDbcUtil müssen Sie nur den spezifischen Datentyp implementieren, den Sie benötigen. dialect.getJDBCType (dt) .orElse (getCommonJDBCType (dt)). getOrElse (neue IllegalArgumentException auslösen (s "JDBC-Typ kann nicht für $ {dt.simpleString} abgerufen werden")
Arnon Rodman
6

Generieren Sie einen einfachen Datensatz mit fünf Werten und konvertieren Sie ihn intin den stringTyp:

val df = spark.range(5).select( col("id").cast("string") )
user8106134
quelle
6

Ich denke, das ist für mich viel besser lesbar.

import org.apache.spark.sql.types._
df.withColumn("year", df("year").cast(IntegerType))

Dadurch wird Ihre Jahresspalte in konvertiert, IntegerTypeindem temporäre Spalten erstellt und diese Spalten gelöscht werden. Wenn Sie in einen anderen Datentyp konvertieren möchten, können Sie die Typen im org.apache.spark.sql.typesPaket überprüfen .

Piyush Patel
quelle
5

Die Antworten, die vorschlagen, Cast, FYI, die Cast-Methode in Spark 1.4.1, zu verwenden, sind fehlerhaft.

Beispielsweise hat ein Datenrahmen mit einer Zeichenfolgenspalte mit dem Wert "8182175552014127960", wenn er in bigint umgewandelt wird, den Wert "8182175552014128100".

    df.show
+-------------------+
|                  a|
+-------------------+
|8182175552014127960|
+-------------------+

    df.selectExpr("cast(a as bigint) a").show
+-------------------+
|                  a|
+-------------------+
|8182175552014128100|
+-------------------+

Wir mussten uns vielen Problemen stellen, bevor wir diesen Fehler fanden, da wir Bigint-Spalten in der Produktion hatten.

sauraI3h
quelle
4
psst, aktualisiere deinen Funken
msemelman
2
@msemelman Es ist lächerlich, für einen kleinen Fehler auf eine neue Version von spark in der Produktion aktualisieren zu müssen.
SauraI3h
Aktualisieren wir nicht immer alles auf kleine Fehler? :)
Caesarsol
5
df.select($"long_col".cast(IntegerType).as("int_col"))
Seelenmaschine
quelle
4

Mit Spark Sql 2.4.0 können Sie Folgendes tun:

spark.sql("SELECT STRING(NULLIF(column,'')) as column_string")
Eric Bellet
quelle
3

Sie können den folgenden Code verwenden.

df.withColumn("year", df("year").cast(IntegerType))

Dadurch wird das Jahr Spalte für IntegerTypeSpalte konvertiert .

adarsh
quelle
2

Diese Methode löscht die alte Spalte und erstellt neue Spalten mit denselben Werten und neuem Datentyp. Meine ursprünglichen Datentypen bei der Erstellung des DataFrame waren: -

root
 |-- id: integer (nullable = true)
 |-- flag1: string (nullable = true)
 |-- flag2: string (nullable = true)
 |-- name: string (nullable = true)
 |-- flag3: string (nullable = true)

Danach habe ich folgenden Code ausgeführt, um den Datentyp zu ändern: -

df=df.withColumnRenamed(<old column name>,<dummy column>) // This was done for both flag1 and flag3
df=df.withColumn(<old column name>,df.col(<dummy column>).cast(<datatype>)).drop(<dummy column>)

Danach war mein Ergebnis: -

root
 |-- id: integer (nullable = true)
 |-- flag2: string (nullable = true)
 |-- name: string (nullable = true)
 |-- flag1: boolean (nullable = true)
 |-- flag3: boolean (nullable = true)
PirateJack
quelle
Könnten Sie bitte hier Ihre Lösung angeben.
Ajay Kharade
1

Sie können den Datentyp einer Spalte ändern, indem Sie cast in spark sql verwenden. Der Tabellenname ist Tabelle und hat nur zwei Spalten. Der Datentyp Spalte1 und Spalte2 und Spalte1 muss geändert werden. ex-spark.sql ("Wählen Sie cast (Spalte1 als Double) column1NewName, column2 aus der Tabelle") Anstelle von double schreiben Sie Ihren Datentyp.

Tejasvi Sharma
quelle
1

Wenn Sie Dutzende von Spalten mit ihrem Namen umbenennen müssen, verwendet das folgende Beispiel den Ansatz von @dnlbrky und wendet ihn auf mehrere Spalten gleichzeitig an:

df.selectExpr(df.columns.map(cn => {
    if (Set("speed", "weight", "height").contains(cn)) s"cast($cn as double) as $cn"
    else if (Set("isActive", "hasDevice").contains(cn)) s"cast($cn as boolean) as $cn"
    else cn
}):_*)

Nicht gegossene Spalten bleiben unverändert. Alle Spalten bleiben in ihrer ursprünglichen Reihenfolge.

kubischer Salat
quelle
1

So viele Antworten und nicht viele gründliche Erklärungen

Die folgende Syntax funktioniert mit Databricks Notebook mit Spark 2.4

from pyspark.sql.functions import *
df = df.withColumn("COL_NAME", to_date(BLDFm["LOAD_DATE"], "MM-dd-yyyy"))

Beachten Sie, dass Sie das Eingabeformat angeben müssen (in meinem Fall "MM-TT-JJJJ") und der Import obligatorisch ist, da to_date eine Spark-SQL-Funktion ist

Versuchte auch diese Syntax, bekam aber Nullen anstelle einer richtigen Besetzung:

df = df.withColumn("COL_NAME", df["COL_NAME"].cast("Date"))

(Hinweis: Ich musste Klammern und Anführungszeichen verwenden, damit es syntaktisch korrekt ist.)


PS: Ich muss zugeben, dass dies wie ein Syntaxdschungel ist, es gibt viele mögliche Einstiegspunkte und den offiziellen API-Referenzen fehlen geeignete Beispiele.

Mehdi LAMRANI
quelle
1
Syntax Dschungel. Ja. Dies ist die Welt von Spark im Moment.
conner.xyz
1

Eine andere Lösung ist wie folgt:

1) Behalten Sie "inferSchema" als falsch bei

2) Während Sie 'Map'-Funktionen in der Zeile ausführen, können Sie' asString 'lesen (row.getString ...)

//Read CSV and create dataset
Dataset<Row> enginesDataSet = sparkSession
            .read()
            .format("com.databricks.spark.csv")
            .option("header", "true")
            .option("inferSchema","false")
            .load(args[0]);

JavaRDD<Box> vertices = enginesDataSet
            .select("BOX","BOX_CD")
            .toJavaRDD()
            .map(new Function<Row, Box>() {
                @Override
                public Box call(Row row) throws Exception {
                    return new Box((String)row.getString(0),(String)row.get(1));
                }
            });
Vibha
quelle
0
    val fact_df = df.select($"data"(30) as "TopicTypeId", $"data"(31) as "TopicId",$"data"(21).cast(FloatType).as( "Data_Value_Std_Err")).rdd
    //Schema to be applied to the table
    val fact_schema = (new StructType).add("TopicTypeId", StringType).add("TopicId", StringType).add("Data_Value_Std_Err", FloatType)

    val fact_table = sqlContext.createDataFrame(fact_df, fact_schema).dropDuplicates()
Aravind Krishnakumar
quelle
0

Ein anderer Weg:

// Generate a simple dataset containing five values and convert int to string type

val df = spark.range(5).select( col("id").cast("string")).withColumnRenamed("id","value")
user8106134
quelle
0

Für den Fall, dass Sie mehrere Spalten eines bestimmten Typs in einen anderen ändern möchten, ohne einzelne Spaltennamen anzugeben

/* Get names of all columns that you want to change type. 
In this example I want to change all columns of type Array to String*/
    val arrColsNames = originalDataFrame.schema.fields.filter(f => f.dataType.isInstanceOf[ArrayType]).map(_.name)

//iterate columns you want to change type and cast to the required type
val updatedDataFrame = arrColsNames.foldLeft(originalDataFrame){(tempDF, colName) => tempDF.withColumn(colName, tempDF.col(colName).cast(DataTypes.StringType))}

//display

updatedDataFrame.show(truncate = false)
Ravi
quelle