Pyspark Dataframe: Проверьте, являются ли значения в столбцах даты действительными - PullRequest
Новая
       11

Pyspark Dataframe: Проверьте, являются ли значения в столбцах даты действительными

0 голосов
/ 27 августа 2018

У меня есть искровой DataFrame, импортированный из файла CSV. После применения некоторых манипуляций (в основном, удаления столбцов / строк) я пытаюсь сохранить новый DataFrame в Hadoop, который выдает сообщение об ошибке:

ValueError: год вне диапазона

Я подозреваю, что некоторые столбцы типа DateType или TimestampType повреждены. По крайней мере, в одном столбце я нашел запись с годом «207» - кажется, это создает проблемы.

** Как я могу проверить, соответствует ли DataFrame требуемым временным диапазонам?

Я думал о написании функции, которая принимает DataFrame и получает для каждого DateType / TimestampType-Column минимальное и максимальное значения, но я не могу заставить это работать. **

Есть идеи?

PS: В моем понимании, spark всегда проверяет и применяет схему. Разве это не включает проверку минимальных / максимальных значений?

1 Ответ

0 голосов
/ 27 августа 2018

Для проверки даты могут помочь регулярные выражения .

например: для проверки даты в формате даты ММ-дд-гггг

step1 : создайте регулярное выражение для вашего формата даты. для MM-дд-гггг это будет ^(0[1-9]|[12][0-9]|3[01])[- \/.](0[1-9]|1[012])[- \/.](19|20)\d\d$

Вы можете использовать этот код для реализации.

Этот шаг поможет найти invalid dates, который не будет анализироваться и вызывать ошибку.

step2 : преобразовать строку в дату. следующий код может помочь 

import scala.util.{Try, Failure}
import org.apache.spark.sql.functions.udf

object FormatChecker extends java.io.Serializable {
  val fmt = org.joda.time.format.DateTimeFormat forPattern "MM-dd-yyyy"
  def invalidFormat(s: String) = Try(fmt parseDateTime s) match {
    case Failure(_) => true
    case _ => false
  }
}

val df = sc.parallelize(Seq(
    "01-02-2015", "99-03-2010", "---", "2015-01-01", "03-30-2001")
).toDF("date")

invalidFormat = udf((s: String) => FormatChecker.invalidFormat(s))
df.where(invalidFormat($"date")).count()
...