Better way to convert a string field into timestamp in Spark

by

Spark >= 2.2

Since you 2.2 you can provide format string directly:

import org.apache.spark.sql.functions.to_timestamp

val ts = to_timestamp($"dts", "MM/dd/yyyy HH:mm:ss")

df.withColumn("ts", ts).show(2, false)

// +---+-------------------+-------------------+
// |id |dts                |ts                 |
// +---+-------------------+-------------------+
// |1  |05/26/2016 01:01:01|2016-05-26 01:01:01|
// |2  |#$@#@#             |null               |
// +---+-------------------+-------------------+

Spark >= 1.6, < 2.2

You can use date processing functions which have been introduced in Spark 1.5. Assuming you have following data:

val df = Seq((1L, "05/26/2016 01:01:01"), (2L, "#$@#@#")).toDF("id", "dts")

You can use unix_timestamp to parse strings and cast it to timestamp

import org.apache.spark.sql.functions.unix_timestamp

val ts = unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss").cast("timestamp")

df.withColumn("ts", ts).show(2, false)

// +---+-------------------+---------------------+
// |id |dts                |ts                   |
// +---+-------------------+---------------------+
// |1  |05/26/2016 01:01:01|2016-05-26 01:01:01.0|
// |2  |#$@#@#             |null                 |
// +---+-------------------+---------------------+

As you can see it covers both parsing and error handling. The format string should be compatible with Java SimpleDateFormat.

Spark >= 1.5, < 1.6

You’ll have to use use something like this:

unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss").cast("double").cast("timestamp")

or

(unix_timestamp($"dts", "MM/dd/yyyy HH:mm:ss") * 1000).cast("timestamp")

due to SPARK-11724.

Spark < 1.5

you should be able to use these with expr and HiveContext.

More Related Contents:

  1. Spark 2.0 Dataset vs DataFrame
  2. Spark – load CSV file as DataFrame?
  3. How to define partitioning of DataFrame?
  4. Encoder error while trying to map dataframe row to updated row
  5. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  6. How to define schema for custom type in Spark SQL?
  7. How can I change column types in Spark SQL’s DataFrame?
  8. Spark unionAll multiple dataframes
  9. How can I pass extra parameters to UDFs in Spark SQL?
  10. How to split a dataframe into dataframes with same column values?
  11. Dropping a nested column from Spark DataFrame
  12. How to define a custom aggregation function to sum a column of Vectors?
  13. Spark DataFrame: does groupBy after orderBy maintain that order?
  14. Filling gaps in timeseries Spark
  15. Spark UDAF with ArrayType as bufferSchema performance issues
  16. Apache Spark how to append new column from list/array to Spark dataframe
  17. How to use COGROUP for large datasets
  18. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  19. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  20. Create new Dataframe with empty/null field values
  21. Perform a typed join in Scala with Spark Datasets
  22. Apache Spark: Get number of records per partition
  23. DataFrame-ified zipWithIndex
  24. Derive multiple columns from a single column in a Spark DataFrame
  25. Spark: Add column to dataframe conditionally
  26. How to save a spark DataFrame as csv on disk?
  27. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  28. Calculate Cosine Similarity Spark Dataframe
  29. Why is join not possible after show operator?
  30. Difference between two rows in Spark dataframe

Leave a Comment