<jupyter_start><jupyter_text>05. Date and time in Spark data framesDate and time column are essential for time series data. In spark, we can express date and time with **four different data types**: - String : Most commun format, no compability issues. But need to be parsed to long or timestamp types to do arithmetic operations- Long : in spark, it represents **unix_timestamp** implementation. it uses seconds since the UNIX epoch- timestamp :  spark has its own implementation to represent timestamp. Represents values comprising values of fields year, month, day, hour, minute, and second, with the **session local time-zone**. The timestamp value represents an absolute point in time.- date: Represents values comprising values of fields year, month and day, without a time-zone.Below, we will show how to do convertion between them<jupyter_code>from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import lit, unix_timestamp, col, from_unixtime, to_timestamp, date_format, to_date
local = True

if local:
    spark=SparkSession.builder.master("local[4]").appName("DateAndTime").getOrCreate()
else:
    spark = SparkSession \
    .builder.master("k8s://https://kubernetes.default.svc:443") \
    .appName("SparkArrowCompression") \
    .config("spark.kubernetes.container.image", "inseefrlab/jupyter-datascience:master") \
    .config("spark.kubernetes.authenticate.driver.serviceAccountName", os.environ['KUBERNETES_SERVICE_ACCOUNT']) \
    .config("spark.executor.instances", "4") \
    .config("spark.executor.memory","8g") \
    .config("spark.kubernetes.namespace", os.environ['KUBERNETES_NAMESPACE']) \
    .getOrCreate()

data = [("2019-07-01 12:01:19", "07-01-2019 12:01:19.888", "07-01-2019"),
            ("2018-07-01 12:01:19", "07-01-2018 12:01:19.666", "07-01-2018"),
            ("2017-07-01 12:01:19", "07-01-2017 12:01:19.111", "07-01-2017")]
columns = ["timestamp_1", "timestamp_2", "timestamp_3"]
df=spark.createDataFrame(data=data,schema=columns)
df.show()
df.printSchema()<jupyter_output>+-------------------+--------------------+-----------+
|        timestamp_1|         timestamp_2|timestamp_3|
+-------------------+--------------------+-----------+
|2019-07-01 12:01:19|07-01-2019 12:01:...| 07-01-2019|
|2018-07-01 12:01:19|07-01-2018 12:01:...| 07-01-2018|
|2017-07-01 12:01:19|07-01-2017 12:01:...| 07-01-2017|
+-------------------+--------------------+-----------+

root
 |-- timestamp_1: string (nullable = true)
 |-- timestamp_2: string (nullable = true)
 |-- timestamp_3: string (nullable = true)<jupyter_text>5.1 Convert string date, timestamp to long (seconds since the UNIX epoch)To convert string to long, we can use unix_timestamp() function, it takes a string date, and a date format, if the date format is not provided, the default format (yyyy-MM-dd HH:mm:ss) will be used. Then the string date are converted to Unix timestamp (in seconds) by using the current timezone of the system. Note you can set the timezone of your spark session by using **spark.conf.set("spark.sql.session.timeZone", "")**, for example spark.conf.set("spark.sql.session.timeZone", "US/Pacific").1: def unix_timestamp(): returns the current time in seconds (LongType)2) def unix_timestamp(date: Column): take the string column date as input, output it in seconds, it will use default date format (yyyy-MM-dd HH:mm:ss) to parse the string. 3) def unix_timestamp(date: Column, format: String): we can specify explicitly the format of the date string, if the default format does not fit.<jupyter_code># We add a new column that is filled with the current time
# note the column type
df1 = df.withColumn("timestamp_4", lit(unix_timestamp()))
df1.printSchema()
df1.show()

# note the output dataframe has timestamp 1 and timestamp 2 correct. But timestamp 3 wrong. 
# we did not give format for timestamp_1 and timestamp_3, it works for timestamp_1 because it uses the default format is yyyy-MM-dd HH:mm:ss
# timestamp_3 does not work, because it does not use the default date format.
# for timestamp_2, we specify a date format, if we don't give the format, it returns null, because the string does not have
# the default date format
# If we give a format that is wrong, the spark job will fail. For example, try to remove SSS from timestamp_2 format, and see what happens

df2 = df.withColumn("timestamp_1", unix_timestamp("timestamp_1")) \
        .withColumn("timestamp_2", unix_timestamp("timestamp_2", "MM-dd-yyyy HH:mm:ss.SSS")) \
        .withColumn("timestamp_3", unix_timestamp("timestamp_3"))
print("Exp1: covert date string to seconds with some erreurs")
df2.printSchema()
df2.show()
# below, we give each one a correct date format.
df3 = df.withColumn("timestamp_1", unix_timestamp("timestamp_1")) \
        .withColumn("timestamp_2", unix_timestamp("timestamp_2", "MM-dd-yyyy HH:mm:ss.SSS")) \
        .withColumn("timestamp_3", unix_timestamp("timestamp_3", "MM-dd-yyyy")) \
        .withColumn("timestamp_4", lit(unix_timestamp()))
print("Exp1: covert date string to seconds with success")
df3.printSchema()
df3.show()<jupyter_output>Exp1: covert date string to seconds with success
root
 |-- timestamp_1: long (nullable = true)
 |-- timestamp_2: long (nullable = true)
 |-- timestamp_3: long (nullable = true)
 |-- timestamp_4: long (nullable = true)

+-----------+-----------+-----------+-----------+
|timestamp_1|timestamp_2|timestamp_3|timestamp_4|
+-----------+-----------+-----------+-----------+
| 1561982479| 1561982479| 1561939200| 1632594849|
| 1530446479| 1530446479| 1530403200| 1632594849|
| 1498910479| 1498910479| 1498867200| 1632594849|
+-----------+-----------+-----------+-----------+<jupyter_text>5.2 Convert long(seconds since the UNIX epoch) back to string dateWe can use from_unixtime(unix_time: Column, format: String) to convert long back to string<jupyter_code># Convert unix timestamp in second to string date
df4 = df3.select(
        from_unixtime(col("timestamp_1")).alias("timestamp_1"),
        from_unixtime(col("timestamp_2"), "MM-dd-yyyy HH:mm:ss").alias("timestamp_2"),
        from_unixtime(col("timestamp_3"), "MM-dd-yyyy").alias("timestamp_3"),
        from_unixtime(col("timestamp_4")).alias("timestamp_4")
    )
df4.printSchema()
df4.show(truncate=False)<jupyter_output>root
 |-- timestamp_1: string (nullable = true)
 |-- timestamp_2: string (nullable = true)
 |-- timestamp_3: string (nullable = true)
 |-- timestamp_4: string (nullable = true)

+-------------------+-------------------+-----------+-------------------+
|timestamp_1        |timestamp_2        |timestamp_3|timestamp_4        |
+-------------------+-------------------+-----------+-------------------+
|2019-07-01 12:01:19|07-01-2019 12:01:19|07-01-2019 |2021-09-25 18:39:03|
|2018-07-01 12:01:19|07-01-2018 12:01:19|07-01-2018 |2021-09-25 18:39:03|
|2017-07-01 12:01:19|07-01-2017 12:01:19|07-01-2017 |2021-09-25 18:39:03|
+-------------------+-------------------+-----------+-------------------+<jupyter_text>5.3 Convert string to spark timestampSpark provides its own timestamp type. Instead of long, we can also use spark timestamp to store time data.To convert a string date to spark timestamp, we can use:- to_timestamp(col, format): the default format is yyyy-MM-dd HH:mm:ss. If it failed, return null.<jupyter_code># to_timestamp() works similar to unix_timestamp(). For timestamp_1, it has the default format, so no need to specify the format
# For 2 and 3, we need to specify the date format. Otherwise, it returns null

# note the output column type

df5 = df.withColumn("timestamp_1", to_timestamp("timestamp_1")) \
        .withColumn("timestamp_2", to_timestamp("timestamp_2", "MM-dd-yyyy HH:mm:ss.SSS")) \
        .withColumn("timestamp_3", to_timestamp("timestamp_3", "MM-dd-yyyy"))

df5.printSchema()
df5.show(truncate=False)
# we can also use to_timestamp in a sql 
spark.sql("select to_timestamp('06-24-2019 12:01:19.000','MM-dd-yyyy HH:mm:ss.SSSS') as timestamp")<jupyter_output><empty_output><jupyter_text>5.4 Convert spark timestamp back to string To convert timestamp back to string, we have two options - cast("String"): It converts a timestamp column to default date format yyyy-MM-dd HH:mm:ss- date_format(column,format): It converts a timestamp column to any Java Date formats specified in DateTimeFormatter.  5.4.1 Use cast("String")<jupyter_code># convert spark timestamp back to string by using cast()
# note they all use the default date format.
df6 = df5.select(col("timestamp_2").cast("string"), \
                col("timestamp_2").cast("string"))
df6.printSchema()
df6.show(truncate=False)<jupyter_output>root
 |-- timestamp_2: string (nullable = true)
 |-- timestamp_2: string (nullable = true)

+-----------------------+-----------------------+
|timestamp_2            |timestamp_2            |
+-----------------------+-----------------------+
|2019-07-01 12:01:19.888|2019-07-01 12:01:19.888|
|2018-07-01 12:01:19.666|2018-07-01 12:01:19.666|
|2017-07-01 12:01:19.111|2017-07-01 12:01:19.111|
+-----------------------+-----------------------+<jupyter_text>5.4.2 Use date_format()<jupyter_code># use date_format() to convert spark timestamp to various string date
df7 = df5.withColumn("str_date_yyyy_MM_dd", date_format(col("timestamp_2"), "yyyy MM dd")) \
        .withColumn("str_date_MM/dd/yyyy_hh:mm", date_format(col("timestamp_2"), "MM/dd/yyyy hh:mm")) \
        .withColumn("str_date_yyyy_MMM_dd", date_format(col("timestamp_2"), "yyyy MMM dd")) \
        .withColumn("str_date_yyyy_MMMM_dd_E", date_format(col("timestamp_2"), "yyyy MMMM dd E"))
    
df7.printSchema()
df7.show(truncate=False)<jupyter_output>root
 |-- timestamp_1: timestamp (nullable = true)
 |-- timestamp_2: timestamp (nullable = true)
 |-- timestamp_3: timestamp (nullable = true)
 |-- str_date_yyyy_MM_dd: string (nullable = true)
 |-- str_date_MM/dd/yyyy_hh:mm: string (nullable = true)
 |-- str_date_yyyy_MMM_dd: string (nullable = true)
 |-- str_date_yyyy_MMMM_dd_E: string (nullable = true)

+-------------------+-----------------------+-------------------+-------------------+-------------------------+--------------------+-----------------------+
|timestamp_1        |timestamp_2            |timestamp_3        |str_date_yyyy_MM_dd|str_date_MM/dd/yyyy_hh:mm|str_date_yyyy_MMM_dd|str_date_yyyy_MMMM_dd_E|
+-------------------+-----------------------+-------------------+-------------------+-------------------------+--------------------+-----------------------+
|2019-07-01 12:01:19|2019-07-01 12:01:19.888|2019-07-01 00:00:00|2019 07 01         |07/01/2019 12:01         |2019 Jul 01         |2019 July 01 Mon       |
|2018-07-01 1[...]<jupyter_text>5.5 use to_date() function to convert string to date type<jupyter_code>data1=[["02-03-2013"], 
       ["05-06-2023"]]
columns1=["input"]
df8 = spark.createDataFrame(data=data1,schema=columns1 )
df9 = df8.withColumn("date", to_date("input", "MM-dd-yyyy"))

df9.printSchema()
df9.show()<jupyter_output>root
 |-- input: string (nullable = true)
 |-- date: date (nullable = true)

+----------+----------+
|     input|      date|
+----------+----------+
|02-03-2013|2013-02-03|
|05-06-2023|2023-05-06|
+----------+----------+
{'fraction_non_alphanumeric': 0.2140935987089833, 'fraction_numerical': 0.0795230410615026, 'mean_word_length': 3.018371757925072, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '145406', 'license': 'MIT', 'n_tokens_mistral': 3998, 'n_tokens_neox': 3426, 'n_words': 947, 'path': 'notebooks/pysparkbasics/L02_DataFrame/S01_DataStructures/05_TimeDate.ipynb', 'repo_name': 'pengfei99/PySparkCommonFunc', 'size': '11154'}