Как прочитать значение столбца dataframe в цикле и проверить тип данных каждого столбца

Question

Мне нужна помощь, чтобы понять итерацию в 'scala'.У меня есть фрейм данных, который имеет другой тип данных (Int, String, Date, Long).Я хочу прочитать данные каждой строки в цикле.если данные столбца, если данные столбца не совпадают с типом данных, то правильный тип данных.Затем я хочу обновить значение столбца с нуля.

Я устал читать каждое значение столбца и проверить с соответствующим типом данных Like 'ID Столбец: целочисленный тип, AirName: строковый тип, Place: строковый тип,TakeoffDate: Date 'type.My Входной набор данных:

+-----+-------+-----+-----------+
|   ID|AirName|Place|TakeoffDate|
|    1|  Delta|  Aus|    1/11/18|
|    2|  Delta|     |    10/5/19|
|Three|   null|  New| 15/10/2018|
|    4| JetAir|  Aus|    11/6/15|
+-----+-------+-----+-----------+

Здесь Three - строковый тип, но ID, который мы объявили как Int.Так что я хочу заменить три на ноль.аналогичный другой столбец в цикле.

Я прочитал данные из файла и создал фрейм данных.Теперь я хочу проверить каждую строку и каждый столбец с соответствующим типом данных.если тип данных не совпадает, я хочу заменить этот столбец на ноль.

Но это не работает для меня.

val myFile = sc.textFile("/FileStore/tables/Airline__2_-14f6c.txt")

import org.apache.spark.sql.types._
case class Test(ID:Int,AirName:String,Place:String,TakeoffDate:String)
val df= myFile.map(x => x.split(",") ).map( x=> Test(x(0).toInt,x(1),x(2),x(3)) ).toDF()

def isInt(in: Integer): Option[Int] = {
    try {
        Some(Integer.parseInt(in))
    } catch {
        case e: NumberFormatException => None
    }
}

rows.map{
     case(ID) => if (isInt(ID).isEmpty (ID,ErrorCodes.NOT_INT)) 
  {
    println("yes")
  }
  else ((Int)ID, None)
  {
    println("no")
  }
}

Expected Output
+-----+-------+-----+-----------+
|   ID|AirName|Place|TakeoffDate|
|    1|  Delta|  Aus|    1/11/18|
|    2|  Delta| null|    10/5/19|
|null |   null|  New|     null  |
|    4| JetAir|  Aus|    11/6/15|
+-----+-------+-----+-----------+

Answer 1

Вместо использования RDD API я бы предложил API dataframe / dataset, который предлагает более богатые функциональные возможности:

import org.apache.spark.sql.functions.{trim, length, when}

val df = Seq(
  ("1", "Delta", "Aus", "1/11/18"),
  ("2", "Delta", null, "10/5/19"),
  ("Three", null, "New", "15/10/2018"),
  ("4", "JetAir", "Aus", "11/6/15"))
.toDF("ID", "AirName","Place", "TakeoffDate")

df.withColumn("ID", $"ID".cast("int"))
.withColumn("TakeoffDate", 
            when(
              $"TakeoffDate".rlike("\\d{1,2}/\\d{1,2}/\\d{1,2}$"), $"TakeoffDate")
            .otherwise(null)
           )
.withColumn("Place", 
            when(
                length(trim($"Place")) > 0, $"Place")
            .otherwise(null))
.show(false)

Вывод:

+----+-------+-----+-----------+
|ID  |AirName|Place|TakeoffDate|
+----+-------+-----+-----------+
|1   |Delta  |Aus  |1/11/18    |
|2   |Delta  |null |10/5/19    |
|null|null   |New  |null       |
|4   |JetAir |Aus  |11/6/15    |
+----+-------+-----+-----------+

Casting

• $"ID".cast("int"): вы приводите любую строку в int.Если преобразование невозможно, приведение вернет ноль по умолчанию.
• $"TakeoffDate".rlike("\\d{1,2}/\\d{1,2}/\\d{1,2}$"): дата должна иметь формат, указанный в этом регулярном выражении.rlike возвращает true, если строка соответствует, в противном случае false.
• length(trim($"Place")) > 0: если строка пуста, возвращает ноль, в противном случае значение $ "Place".