Załóżmy, że robię coś takiego:

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...

Ale naprawdę chciałem yearas Int(i być może przekształcić kilka innych kolumn).

Najlepsze, co mogłem wymyślić, to

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]

co jest nieco zawiłe.

Pochodzę z R i jestem przyzwyczajony do pisania np

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

Prawdopodobnie czegoś mi brakuje, ponieważ powinien być lepszy sposób na zrobienie tego w Spark / Scala ...

Edycja: najnowsza wersja

Od Spark 2.x możesz używać .withColumn. Sprawdź dokumenty tutaj:

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

Najstarsza odpowiedź

Od wersji Spark 1.4 możesz zastosować metodę rzutowania z DataType w kolumnie:

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

Jeśli używasz wyrażeń sql, możesz również wykonać:

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

Więcej informacji można znaleźć w dokumentacji: http://spark.apache.org/docs/1.6.0/api/scala/#org.apache.spark.sql.DataFrame

[EDYCJA: marzec 2016 r .: dzięki za głosy! Choć tak naprawdę, to nie jest najlepsza odpowiedź, myślę, że rozwiązania oparte na withColumn, withColumnRenameda castpodniesione przez msemelman Martin Senne i inni są prostsze i czystsze].

Myślę, że twoje podejście jest w porządku, pamiętaj, że Spark DataFramejest (niezmiennym) RDD wierszy, więc tak naprawdę nigdy nie zastępujemy kolumny, po prostu tworzymy nowy za DataFramekażdym razem z nowym schematem.

Zakładając, że masz oryginalny plik df z następującym schematem:

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)

I niektóre UDF zdefiniowane w jednej lub kilku kolumnach:

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
 )

Zmiana typów kolumn, a nawet tworzenie nowej ramki DataFrame z innej, można zapisać w następujący sposób:

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")            

co daje:

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)

Jest to bardzo zbliżone do twojego własnego rozwiązania. Po prostu zachowanie zmian typu i innych przekształceń jako oddzielnych udf valsprawia, że ​​kod jest bardziej czytelny i można go ponownie wykorzystać.

Ponieważ castoperacja jest dostępna dla Sparka Column(i osobiście nie popieram udfproponowanej przez @ Svendw tym momencie), co powiesz na:

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

rzutować na żądany typ? Jako fajny efekt uboczny, wartości, których nie można rzutować / „konwertować” w tym sensie, staną się null.

Jeśli potrzebujesz tego jako metody pomocniczej , użyj:

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

który jest używany jak:

import DFHelper._
val df2 = castColumnTo( df, "year", IntegerType )

Po pierwsze , jeśli chcesz przesyłać typ, to:

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

Kolumna o tej samej nazwie zostanie zastąpiona nową. Nie musisz dodawać ani usuwać kroków.

Po drugie , o Scala vs R .
Oto kod, który najbardziej przypomina RI:

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)
   }: _*
)

Chociaż długość kodu jest nieco dłuższa niż R. Nie ma to nic wspólnego z gadatliwością języka. W R mutatejest to specjalna funkcja dla R dataframe, natomiast w Scali można ją łatwo ad-hoc dzięki jej ekspresyjnej sile.
Słowem, unika konkretnych rozwiązań, ponieważ projekt języka jest wystarczająco dobry, abyś mógł szybko i łatwo zbudować własny język domeny.

uwaga boczna: df.columnsjest zaskakująco a Array[String]zamiast Array[Column], może chcą, aby wyglądało jak ramka danych pandy w Pythonie.

Możesz użyć, selectExpraby uczynić go trochę czystszym:

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

Kod Java służący do modyfikowania typu danych DataFrame z String na Integer

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

Po prostu przerzuci istniejący (typ danych String) na liczbę całkowitą.

Aby przekonwertować rok ze string na int, możesz dodać następującą opcję do czytnika csv: "inferSchema" -> "true", zobacz dokumentację DataBricks

Więc to naprawdę działa tylko wtedy, gdy masz problemy z zapisywaniem do sterownika jdbc, takiego jak sqlserver, ale jest naprawdę pomocne w przypadku błędów, które napotkasz ze składnią i typami.

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)

Wygeneruj prosty zbiór danych zawierający pięć wartości i przekonwertuj intna stringtyp:

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

Myślę, że jest to dla mnie dużo bardziej czytelne.

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

Spowoduje to konwersję kolumny roku na IntegerTypez utworzeniem jakichkolwiek kolumn tymczasowych i upuszczeniem tych kolumn. Jeśli chcesz przekonwertować na inny typ danych, możesz sprawdzić typy w org.apache.spark.sql.typespakiecie.

odpowiedzi sugerujące użycie cast, FYI, metoda cast w Spark 1.4.1 jest zepsuta.

na przykład ramka danych z kolumną łańcuchową o wartości „8182175552014127960” po rzutowaniu na bigint ma wartość „8182175552014128100”

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

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

Musieliśmy zmierzyć się z wieloma problemami, zanim znaleźliśmy ten błąd, ponieważ mieliśmy kolumny bigint w produkcji.

df.select($"long_col".cast(IntegerType).as("int_col"))

Używając Spark Sql 2.4.0 możesz to zrobić:

spark.sql("SELECT STRING(NULLIF(column,'')) as column_string")

Możesz użyć poniższego kodu.

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

Który przekształci kolumnę roku w IntegerTypekolumnę.

Ta metoda spowoduje usunięcie starej kolumny i utworzenie nowych kolumn z tymi samymi wartościami i nowym typem danych. Moje oryginalne typy danych podczas tworzenia DataFrame to: -

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

Po tym uruchomiłem następujący kod, aby zmienić typ danych: -

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>)

Po tym mój wynik wyszedł: -

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

Można zmienić typ danych kolumny za pomocą rzutowania w Spark sql. nazwa tabeli to tabela i ma dwie kolumny, tylko typ danych kolumna1 i kolumna2, a typ danych kolumna1 ma zostać zmieniony. ex-spark.sql ("select cast (column1 as Double) column1NewName, column2 from table") W miejsce double zapisz swój typ danych.

W przypadku, gdy musisz zmienić nazwę dziesiątek kolumn podanych przez ich nazwę, poniższy przykład przyjmuje podejście @dnlbrky i stosuje je do kilku kolumn jednocześnie:

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
}):_*)

Nieobrzucone kolumny pozostają niezmienione. Wszystkie kolumny pozostają w pierwotnej kolejności.

Tyle odpowiedzi i niewiele dokładnych wyjaśnień

Następująca składnia działa przy użyciu Notatnika Databricks ze Spark 2.4

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

Zauważ, że musisz określić format wpisu jaki posiadasz (w moim przypadku "MM-dd-rrrr"), a import jest obowiązkowy, ponieważ to_date jest funkcją iskrową sql

Wypróbowałem również tę składnię, ale otrzymałem wartości null zamiast prawidłowego rzutowania:

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

(Uwaga, musiałem użyć nawiasów i cudzysłowów, aby było to poprawne składniowo)


PS: Muszę przyznać, że to jest jak dżungla składni, istnieje wiele możliwych sposobów wejścia, a oficjalne odniesienia do API nie mają odpowiednich przykładów.

Inne rozwiązanie jest następujące:

1) Zachowaj „inferSchema” jako False

2) Podczas uruchamiania funkcji „Map” w wierszu można odczytać „asString” (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));
                }
            });

Dlaczego nie zrobić tak, jak opisano w http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.Column.cast

df.select(df.year.cast("int"),"make","model","comment","blank")

    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()

Inny sposób:

// 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")

W przypadku, gdy chcesz zmienić wiele kolumn określonego typu na inne bez określania nazw poszczególnych kolumn

/* 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)