Comment changer le type de données d'une colonne dans StructField d'un StructType?

Pour modifier un schéma DataFrame spécifique à un type de données donné, vous pouvez faire correspondre un modèle avec dataType de StructField , comme indiqué ci-dessous:

val pattern = """DecimalType\(\d+,(\d+)\)""".r

val df2 = df.dtypes.
  collect{ case (dn, dt) if pattern.findFirstMatchIn(dt).map(_.group(1)).getOrElse("0") != "0" => dn }.
  foldLeft(df)((accDF, c) => accDF.withColumn(c, col(c).cast("Double")))

Cependant, en supposant que votre objectif final est de transformer l'ensemble de données avec le changement de type de colonne, il serait plus facile de simplement parcourir les colonnes du type de données ciblé pour les cast de manière itérative, comme ci-dessous: p >

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

val df2 = df.dtypes.
  collect{ case (dn, dt) if dt.startsWith("DecimalType") => dn }.
  foldLeft(df)((accDF, c) => accDF.withColumn(c, col(c).cast("Double")))

df2.printSchema
// root
//  |-- c1: long (nullable = false)
//  |-- c2: double (nullable = true)
//  |-- c3: string (nullable = true)
//  |-- c4: double (nullable = true)

[UPDATE]

Par exigence supplémentaire des commentaires, si vous souhaitez modifier le schéma uniquement pour DecimalType avec une échelle positive, il suffit d'appliquer une correspondance de motif Regex comme condition de guard dans la méthode collect:

import org.apache.spark.sql.types._

val df = Seq(
  (1L, BigDecimal(12.34), "a", BigDecimal(10.001)),
  (2L, BigDecimal(56.78), "b", BigDecimal(20.002))
).toDF("c1", "c2", "c3", "c4")

val newSchema = df.schema.fields.map{
  case StructField(name, _: DecimalType, nullable, _)
    => StructField(name, DoubleType, nullable)
  case field => field
}
// newSchema: Array[org.apache.spark.sql.types.StructField] = Array(
//   StructField(c1,LongType,false), StructField(c2,DoubleType,true),
//   StructField(c3,StringType,true), StructField(c4,DoubleType,true)
// )

Voici une autre façon:

val newDF = data.selectExpr(newSchema.map(
   col => s"CAST ( ${col.name} As ${col.dataType.sql}) ${col.name}"
  ): _*)

newDF.printSchema

root
 |-- col1: string (nullable = false)
 |-- col2: double (nullable = true)
 |-- col3: string (nullable = true)
 |-- col4: double (nullable = true) 

newDF.show(false)
+----+-----------+----+------+
|col1|col2       |col3|col4  |
+----+-----------+----+------+
|1   |0.0032     |a   |23.32 |
|2   |78787.99003|c   |343.32|
+----+-----------+----+------+

Créez un schéma que vous voulez:
Exemple:

val newSchema = StructType(
  Seq(
    StructField("col1", StringType, true),
    StructField("col2", DoubleType, true),
    StructField("col3", StringType, true),
    StructField("col4", DoubleType, true)
  )
)

Convertissez les colonnes dans le type de données requis.

data.show(false)
data.printSchema

+----+------------------------+----+----------------------+
|col1|col2                    |col3|col4                  |
+----+------------------------+----+----------------------+
|1   |0.003200000000000000    |a   |23.320000000000000000 |
|2   |78787.990030000000000000|c   |343.320000000000000000|
+----+------------------------+----+----------------------+

root
 |-- col1: integer (nullable = false)
 |-- col2: decimal(38,18) (nullable = true)
 |-- col3: string (nullable = true)
 |-- col4: decimal(38,18) (nullable = true) 

val transformedColumns = inputDataDF.dtypes
      .collect {
        case (dn, dt)
            if (dt.startsWith("DecimalType")) =>
          (dn, DoubleType)
      }

    val transformedDF = inputDataDF.select(transformedColumns.map { fieldType =>
      inputDataDF(fieldType._1).cast(fieldType._2)
    }: _*)

vous pouvez en savoir plus sur le coût de withColumn ici - https: //medium.com/@manuzhang/the-hidden-cost-of-spark-withcolumn-8ffea517c015