Cast column containing multiple string date formats to DateTime in Spark

by

Personally I would recommend using SQL functions directly without expensive and inefficient reformatting:

from pyspark.sql.functions import coalesce, to_date

def to_date_(col, formats=("MM/dd/yyyy", "yyyy-MM-dd")):
    # Spark 2.2 or later syntax, for < 2.2 use unix_timestamp and cast
    return coalesce(*[to_date(col, f) for f in formats])

This will choose the first format, which can successfully parse input string.

Usage:

df = spark.createDataFrame([(1, "01/22/2010"), (2, "2018-12-01")], ("id", "dt"))
df.withColumn("pdt", to_date_("dt")).show()

+---+----------+----------+
| id|        dt|       pdt|
+---+----------+----------+
|  1|01/22/2010|2010-01-22|
|  2|2018-12-01|2018-12-01|
+---+----------+----------+

It will be faster than udf, and adding new formats is just a matter of adjusting formats parameter.

However it won’t help you with format ambiguities. In general case it might not be possible to do it without manual intervention and cross referencing with external data.

The same thing can be of course done in Scala:

import org.apache.spark.sql.Column
import org.apache.spark.sql.functions.{coalesce, to_date}

def to_date_(col: Column, 
             formats: Seq[String] = Seq("MM/dd/yyyy", "yyyy-MM-dd")) = {
  coalesce(formats.map(f => to_date(col, f)): _*)
}

More Related Contents:

  1. Spark Equivalent of IF Then ELSE
  2. How to add a constant column in a Spark DataFrame?
  3. Spark Dataframe distinguish columns with duplicated name
  4. How do I add a new column to a Spark DataFrame (using PySpark)?
  5. How to perform union on two DataFrames with different amounts of columns in spark?
  6. Does spark predicate pushdown work with JDBC?
  7. How to change dataframe column names in pyspark?
  8. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  9. Explode in PySpark
  10. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  11. Passing a data frame column and external list to udf under withColumn
  12. Updating a dataframe column in spark
  13. Dividing complex rows of dataframe to simple rows in Pyspark
  14. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  15. Why is Apache-Spark – Python so slow locally as compared to pandas?
  16. Spark add new column to dataframe with value from previous row
  17. PySpark create new column with mapping from a dict
  18. Pyspark: Replacing value in a column by searching a dictionary
  19. Create Spark DataFrame. Can not infer schema for type
  20. Pivot String column on Pyspark Dataframe
  21. Using UDF ignores condition in when
  22. Create single row dataframe from list of list PySpark
  23. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  24. Concatenate two PySpark dataframes
  25. Multiple condition filter on dataframe
  26. How to return a “Tuple type” in a UDF in PySpark?
  27. How to connect HBase and Spark using Python?
  28. How to pivot on multiple columns in Spark SQL?
  29. Apache Spark — Assign the result of UDF to multiple dataframe columns
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment