Ответ 1

Изменение: новейшая версия

Начиная с spark 2.x вы можете использовать .withColumn. Проверьте документы здесь:

https://spark.apache.org/docs/latest/api/scala/index.html#[email protected](colName:String,col:org.apache.spark.sql.Column): org.apache.spark.sql.DataFrame

Самый старый ответ

Начиная с версии Spark 1.4 вы можете применить метод cast с DataType к столбцу:

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

Если вы используете выражения SQL, вы также можете сделать:

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

Для получения дополнительной информации проверьте документы: http://spark.apache.org/docs/1.6.0/api/scala/#org.apache.spark.sql.DataFrame

Ответ 2

[EDIT: март 2016: спасибо за голоса! Хотя на самом деле это не лучший ответ, я думаю, что решения, основанные на withColumn, withColumnRenamed и cast, выдвинутые msemelman, Martin Senne и другими, проще и чище].

Я думаю, что ваш подход в порядке, напомним, что Spark DataFrame является (неизменным) RDD строк, поэтому мы никогда не заменяем столбец, просто создавая новый DataFrame каждый раз с новой схемой.

Предполагая, что у вас есть исходный df со следующей схемой:

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)

И некоторые UDF определены в одном или нескольких столбцах:

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
 )

Изменение типов столбцов или даже создание нового DataFrame из другого можно записать следующим образом:

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

который дает:

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)

Это довольно близко к вашему собственному решению. Просто, сохраняя изменения типа и другие преобразования как отдельные udf val, сделать код более читаемым и повторно использоваться.

Ответ 3

Поскольку операция cast доступна для Spark Column (и поскольку я лично не одобряю udf, как было предложено @Svend в этой точке), как насчет:

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

чтобы применить к запрашиваемому типу? В качестве аккуратного побочного эффекта значения, не зависящие/ "конвертируемые" в этом смысле, станут null.

Если вам нужно это как вспомогательный метод, используйте:

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

который используется как:

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

Ответ 4

Сначала, если вы хотите разыграть тип, то это:

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

С тем же именем столбца столбец будет заменен новым. Вам не нужно добавлять и удалять шаги.

Второй о Скале против Р.
Это код, который больше всего похож на R, который я могу придумать:

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

Хотя длина кода немного больше, чем у R. Это не имеет ничего общего с многословием языка. В R mutate - это специальная функция для R-данных, в то время как в Scala вы можете легко использовать ее благодаря своей выразительной мощности.
Словом, он избегает конкретных решений, потому что фундамент достаточно хорош, чтобы вы могли быстро и легко создавать свои собственные функции языка домена.

примечание: df.columns на удивление - Array[String] вместо Array[Column], возможно, они хотят, чтобы он выглядел как датафрейм Python-панд.

Ответ 5

Вы можете использовать selectExpr, чтобы сделать его немного чище:

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

Ответ 6

Java-код для изменения типа данных DataFrame от String до Integer

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

Он просто преобразует существующий (тип данных String) в Integer.

Ответ 7

Чтобы преобразовать год из строки в int, вы можете добавить в csv-считыватель следующую опцию: "inferSchema" → "true", см. Документация DataBricks

Ответ 8

Итак, это действительно работает, если у вас есть проблемы с сохранением драйвера jdbc, такого как sqlserver, но он действительно полезен для ошибок, с которыми вы столкнетесь с синтаксисом и типами.

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)

Ответ 9

Создайте простой набор данных, содержащий пять значений, и преобразуйте int в string тип:

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

Ответ 10

ответы, предлагающие использовать литые, FYI, метод литья в искре 1.4.1 нарушен.

например, фреймворк данных со строковым столбцом, имеющим значение "8182175552014127960" при передаче в bigint, имеет значение "8182175552014128100"

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

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

Нам пришлось столкнуться с большим количеством проблем, прежде чем найти эту ошибку, потому что у нас были большие столбцы в производстве.

Ответ 11

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

Ответ 12

Используя Spark Sql 2.4.0, вы можете сделать это:

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

Ответ 13

Этот метод потеряет старый столбец и создаст новые столбцы с одинаковыми значениями и новым типом данных. Мои исходные типы данных при создании DataFrame были: -

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

После этого я выполнил следующий код, чтобы изменить тип данных: -

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

После этого мой результат получился: -

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

Ответ 14

Вы можете использовать приведенный ниже код.

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

Который будет конвертировать столбец год в столбец IntegerType.

Ответ 15

Если вам нужно переименовать десятки столбцов, заданных их именами, в следующем примере используется подход @dnlbrky и он применяется к нескольким столбцам одновременно:

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

Неклассированные столбцы остаются без изменений. Все столбцы остаются в исходном порядке.

Ответ 16

Можно изменить тип данных столбца, используя метод cast in spark sql. имя таблицы - это таблица, и в ней есть только два столбца: только столбцы column1 и column2 и тип данных столбца1 должны быть изменены. ex-spark.sql( "select cast (column1 as Double) column1NewName, column2 из таблицы" ) Вместо двойной записи введите свой тип данных.

Ответ 17

Another solution is as follows:
1) Keep "inferSchema" as False
2) While running 'Map' functions on the row, you can read 'asString' (row.getString...)

<Code>
        //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));
                        }
                    });
</Code>

Ответ 18

Я думаю, что это намного более читабельно для меня.

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

Это преобразует ваш столбец года в IntegerType с созданием любых временных столбцов и удалением этих столбцов. Если вы хотите преобразовать в любой другой тип данных, вы можете проверить типы внутри пакета org.apache.spark.sql.types.

Ответ 19

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

Ответ 20

По-другому:

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