Date and Time Functions- Pyspark Dataframes & Pyspark Sql Queries

How to handle string Datetime variables in pyspark?

Handling string-formatted datetime variables in PySpark requires transforming them into PySpark-compatible datetime types (DateType or TimestampType) for effective processing. Below is a consolidated guide to handle string datetime variables in PySpark efficiently:

---

1. Convert String Datetime to PySpark Date/Time Types

Steps:

1. Use to_date to convert a string to DateType (YYYY-MM-DD format).
2. Use to_timestamp to convert a string to TimestampType (YYYY-MM-DD HH:mm:ss format).

from pyspark.sql.functions import to_date, to_timestamp, col

# Sample DataFrame
data = [("2024-01-15", "2024-01-15 12:34:56")]
columns = ["string_date", "string_datetime"]
df = spark.createDataFrame(data, columns)

# Convert string to DateType
df = df.withColumn("date_col", to_date(col("string_date"), "yyyy-MM-dd"))

# Convert string to TimestampType
df = df.withColumn("timestamp_col", to_timestamp(col("string_datetime"), "yyyy-MM-dd HH:mm:ss"))

df.show()

Output:

diffCopy code+-----------+-------------------+----------+-------------------+
|string_date|string_datetime    |date_col  |timestamp_col      |
+-----------+-------------------+----------+-------------------+
|2024-01-15 |2024-01-15 12:34:56|2024-01-15|2024-01-15 12:34:56|
+-----------+-------------------+----------+-------------------+

=========================================

’12-Feb-2024′ is a string How to convert it into a date Value in pyspark?

We can use the to_date() function in PySpark to convert a string to a date. Here’s an example:

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_date, col

# create a SparkSession
spark = SparkSession.builder.appName("Date Conversion").getOrCreate()

# create a sample DataFrame
data = [("12-Feb-2024",)]
df = spark.createDataFrame(data, ["date_str"])

# convert the string to a date
df = df.withColumn("date", to_date(col("date_str"), "dd-MMM-yyyy"))

# show the result
df.show()

In this example, the to_date() function takes two arguments: the column to convert (date_str) and the format of the date string ("dd-MMM-yyyy"). The format string tells PySpark how to parse the date string.

The output of this code will be:

+-------------+----------+
|     date_str|      date|
+-------------+----------+
|12-Feb-2024|2024-02-12|
+-------------+----------+

Note that the resulting date column is in the format yyyy-MM-dd, which is the default format for dates in PySpark.

=============================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, to_timestamp

# Initialize Spark session
spark = SparkSession.builder.master("local").appName("Dynamic Formatting").getOrCreate()

# Sample data
data = [("01/15/2024", "15-Jan-2024 12:34:56")]
columns = ["date_format1", "datetime_format2"]
df = spark.createDataFrame(data, columns)

# Define formats in a dictionary
date_formats = {
    "date_format1": "MM/dd/yyyy",
    "datetime_format2": "dd-MMM-yyyy HH:mm:ss"
}

# Use the formats dynamically with f-strings
df = df.withColumn("parsed_date", to_date(col("date_format1"), f"{date_formats['date_format1']}")) 
       .withColumn("parsed_timestamp", to_timestamp(col("datetime_format2"), f"{date_formats['datetime_format2']}"))

# Show the resulting DataFrame
df.show(truncate=False)

---

2. Extract Components from Datetime

Extract useful components like year, month, day, hour, etc.

from pyspark.sql.functions import year, month, dayofmonth, hour, minute, second

df = df.withColumn("year", year(col("timestamp_col"))) 
       .withColumn("month", month(col("timestamp_col"))) 
       .withColumn("day", dayofmonth(col("timestamp_col"))) 
       .withColumn("hour", hour(col("timestamp_col"))) 
       .withColumn("minute", minute(col("timestamp_col"))) 
       .withColumn("second", second(col("timestamp_col")))

df.show()

---

3. Filter Data Based on String DateTime

For string-based filtering, PySpark allows direct comparison without converting strings. For better accuracy, convert to date/time first.

Filter Examples:

# Filter by exact date
filtered_df = df.filter(col("string_date") == "2024-01-15")

# Filter by range (convert to date first for accuracy)
filtered_df = df.filter((col("date_col") >= "2024-01-01") & (col("date_col") <= "2024-12-31"))

filtered_df.show()

---

4. Format Dates into Custom Strings

Convert DateType or TimestampType into custom string formats.

from pyspark.sql.functions import date_format

# Format Timestamp into a custom string
df = df.withColumn("custom_format", date_format(col("timestamp_col"), "dd-MM-yyyy HH:mm:ss"))

df.show()

---

5. Handle Different String Formats

If datetime strings come in various formats, specify the exact format during conversion.

Example:

data = [("01/15/2024", "15-Jan-2024 12:34:56")]
columns = ["date_format1", "datetime_format2"]
df = spark.createDataFrame(data, columns)

df = df.withColumn("parsed_date", to_date(col("date_format1"), "MM/dd/yyyy")) 
       .withColumn("parsed_timestamp", to_timestamp(col("datetime_format2"), "dd-MMM-yyyy HH:mm:ss"))

df.show()

---

6. Handle Null or Invalid Datetime Strings

PySpark handles invalid datetime strings as null during conversion. Use na.drop or na.fill to manage nulls.

Example:

data = [("2024-01-15", "invalid_date")]
columns = ["valid_date", "invalid_date"]
df = spark.createDataFrame(data, columns)

# Attempt to convert invalid strings
df = df.withColumn("converted_date", to_date(col("invalid_date"), "yyyy-MM-dd"))

# Handle nulls
df = df.na.fill({"converted_date": "1970-01-01"})  # Fill nulls with default date

df.show()

---

7. Aggregate Data Based on Datetime

Group data by date/time components or custom time intervals.

Example: Group by Year-Month:

df = df.withColumn("year_month", date_format(col("date_col"), "yyyy-MM"))
aggregated_df = df.groupBy("year_month").count()

aggregated_df.show()

---

8. Compare Dates Across Different Formats

If comparing string-formatted Oracle datetime with PySpark date:

1. Convert Oracle string to PySpark date.
2. Compare the two dates.

Example:

data = [("2024-01-15 12:34:56", "2024-01-15")]
columns = ["oracle_datetime", "hive_date"]
df = spark.createDataFrame(data, columns)

df = df.withColumn("parsed_oracle_date", to_date(col("oracle_datetime"), "yyyy-MM-dd HH:mm:ss")) 
       .withColumn("comparison", col("parsed_oracle_date") == col("hive_date"))

df.show()

---

Best Practices:

1. Use Native Functions: Avoid Python UDFs to ensure Catalyst optimization.
2. Validate Formats: Ensure all string datetime formats are pre-validated.
3. Handle Nulls Gracefully: Use na.fill or na.drop where applicable.
4. Leverage Partitioning: For ETL pipelines, partition large datasets by year/month/day for better performance.

These methods ensure robust and efficient handling of string datetime variables in PySpark for ETL processes.

Question-A table has a column program_date column which has datetime values in string format , it has to be compared with another table which has colum month_year whose values can be in integer( yyyymm) 201402 or string (yyyymmm) i.e 2014Feb, How can it be done in Pyspark Dataframe Api or Spark SQL?

We can achieve this by performing the following steps:

Step 1: Convert the program_date column to a date format

Use the to_date function to convert the program_date column to a date format.

from pyspark.sql.functions import to_date

df1 = df1.withColumn("program_date", to_date("program_date", "dd-MMM-yyyy"))

Step 2: Extract the year and month from the program_date column

Use the year and month functions to extract the year and month from the program_date column.

from pyspark.sql.functions import year, month

df1 = df1.withColumn("year", year("program_date")) 
          .withColumn("month", month("program_date"))

Step 3: Convert the month_year column in the second table to a consistent format

Use the when and otherwise functions to check if the month_year column is in integer or string format, and convert it accordingly.

from pyspark.sql.functions import when, otherwise, substr, length

df2 = df2.withColumn("month_year", 
                       when(length("month_year") == 6, "month_year") 
                       .otherwise(substr("month_year", 1, 4) * 100 + 
                                    when(substr("month_year", 5, 3) == "Jan", 1)
                                    .when(substr("month_year", 5, 3) == "Feb", 2)
                                    .when(substr("month_year", 5, 3) == "Mar", 3)
                                    .when(substr("month_year", 5, 3) == "Apr", 4)
                                    .when(substr("month_year", 5, 3) == "May", 5)
                                    .when(substr("month_year", 5, 3) == "Jun", 6)
                                    .when(substr("month_year", 5, 3) == "Jul", 7)
                                    .when(substr("month_year", 5, 3) == "Aug", 8)
                                    .when(substr("month_year", 5, 3) == "Sep", 9)
                                    .when(substr("month_year", 5, 3) == "Oct", 10)
                                    .when(substr("month_year", 5, 3) == "Nov", 11)
                                    .when(substr("month_year", 5, 3) == "Dec", 12)
                                    .otherwise(0)))

Step 4: Join the two tables based on the year and month columns

Use the join function to join the two tables based on the year and month columns.

df_joined = df1.join(df2, (df1.year * 100 + df1.month) == df2.month_year)

---

We can achieve this in Spark SQL by using a combination of the unix_timestamp function to convert the string datetime to a timestamp, and the year and month functions to extract the year and month from the timestamp. Then, you can format the month_year column to match the desired format.

Here’s an example Spark SQL query:

SELECT *
FROM table1
JOIN table2
ON (
  (YEAR(FROM_UNIXTIME(UNIX_TIMESTAMP(table1.program_date, 'dd-MMM-yyyy'), 'yyyy-MM-dd')) * 100) + 
  MONTH(FROM_UNIXTIME(UNIX_TIMESTAMP(table1.program_date, 'dd-MMM-yyyy'), 'yyyy-MM-dd'))
) = 
CASE 
  WHEN table2.month_year LIKE '%[^0-9]%' THEN 
    CAST(SUBSTR(table2.month_year, 1, 4) AS INT) * 100 + 
    CASE 
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'jan' THEN 1
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'feb' THEN 2
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'mar' THEN 3
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'apr' THEN 4
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'may' THEN 5
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'jun' THEN 6
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'jul' THEN 7
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'aug' THEN 8
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'sep' THEN 9
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'oct' THEN 10
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'nov' THEN 11
      WHEN LOWER(SUBSTR(table2.month_year, 5, 3)) = 'dec' THEN 12
    END
  ELSE CAST(table2.month_year AS INT)
END