How can I let PySpark recognize a column as a datetime type?

Question

I'm using SparkSession.createDataFrame to create a Dataframe from a list of dict like this:

data=[
    {
        'id':1,
        'create_time':datetime.datetime('2022','9','9','0','0','0')
    },
    {
        'id':2,
        'create_time':datetime.datetime('2022','9','9','0','0','0')
    }
]

dataframe = sparkSession.createDataFrame(data)

But Spark raises an exception:

pyspark.sql.utils.AnalysisException: cannot resolve 'create_time' given input columns

Is this because PySpark can not resolve the datetime.datetime type? How should I convert the value of 'create_time' in order to let Spark recognize this column as datetime type?

MaKaNu · Answer

As the comments already mentioned: Use Integer for datetime:

data=[
    {
        'id':1,
        'create_time':datetime.datetime(2022,9,9,0,0,0)
    },
    {
        'id':2,
        'create_time':datetime.datetime(2023,9,9,0,0,0)
    }
]

dataframe = spark.createDataFrame(data)

I recommend here to follow the official documentation and use Spark for the SparkSession to work on the same variable naming.

Further to your question in the comments:

If you inspect your dataframe,

print(dataframe)

>>>DataFrame[create_time: timestamp, id: bigint]

you may notice, that create_time, as well as id, got a type. This is reasonable, because every data item has a need for a datatype. In Python, datatypes are provided dynamically. I assume here (I am not completely into Spark) that the Spark dataframe uses static datatypes. So even that you didn't specify the type for the column id, as soon as you use the createDataFrame method, the type will be determined based on the datatype of the number variables type at this specific moment. So basically if I use

data=[
    {
        'id':1.0,
        // ...

    },
    {
        'id':2.0,
        // ...
    }
]

it will not be represented as bigint, but as double. If you try to mix the types, like first as double and second as bigint, you will be presented with this nice error message:

TypeError: field id: Can not merge type  and

This somehow prove my assumption about static types.

So even as you don't want to use a schema, Spark will determine the schema based on your data inputs as

dataframe.printSchema()
dataframe.show()
>>>root
    |-- create_time: timestamp (nullable = true)
    |-- id: double (nullable = true)

>>>+-------------------+---+
   |        create_time| id|
   +-------------------+---+
   |2022-09-09 00:00:00|  1|
   |2022-09-09 00:00:00|  2|
   +-------------------+---+

will show.

CRAFTY DBA · Answer

To solve this problem, we need to know about list, tuples, and data types. This is key to create the Python structure that is converted into a dataframe. However, inferring versus defining a schema is equally important.

First, I am going to create a dataframe from two tuples. The first field is an integer and the second field is a string. I am supplying both the data and columns as parameters. In this case, Spark is inferring the data.

#
# 1 - Create sample dataframe + view
#

# array of tuples - data
dat1 = [
  (1, "2022-09-09T14:00:00"),
  (2, "2022-09-09T16:00:00")
]

# array of names - columns
col1 = ["event_id", "event_start"]

# make data frame
df1 = spark.createDataFrame(data=dat1, schema=col1)

# make temp hive view
df1.createOrReplaceTempView("event_data1")

# show schema
df1.printSchema()

The screen below shows the data is formatted as a number and a string within our source list. Since we just passed column names without any schema definition to the create data frame method, the resulting data types are inferred. The resulting dataframe has a long and string data types for the columns.

Second, we can not only change the data type within the source list, but we can also supply a schema. Supplying a schema is key for large ASCII formats, such as CSV, JSON, and XML. This stops the Spark engine from reading the whole file to infer the data type.

#
# 2 - Create sample dataframe + view
#

from datetime import datetime
from pyspark.sql.types import *

# array of tuples - data
dat2 = [
  (1, datetime.strptime('2022-09-09 14:00:00',  '%Y-%m-%d %H:%M:%S') ),
  (2, datetime.strptime('2022-09-09 16:00:00', '%Y-%m-%d %H:%M:%S') )
]

# array of names - columns
col2 = StructType([
   StructField("event_id", IntegerType(), True),
   StructField("event_start", TimestampType(), True)])

# make data frame
df2 = spark.createDataFrame(data=dat2, schema=col2)

# make temp hive view
df2.createOrReplaceTempView("event_data2")

# show schema
df2.printSchema()

The image below shows we now have an integer and timestamp data types for both the list and dataframe.

Sometimes, data is problematic in nature. Therefore, we want to import the data as a string and then apply a conversion function.

Third, the conversion of the data afterwards handles malformed data quite well.

#
# 3 - Create sample dataframe + view
#

from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from pyspark.sql.functions import *

# array of tuples - data
dat3 = [
#  (1, '2022-09-09 14:00:00'),
  (1, '2'),
  (2, '2022-09-09 16:00:00')
]

# array of names - columns
col3 = StructType([
   StructField("event_id", IntegerType(), True),
   StructField("event_start", StringType(), True)])

# make data frame
df3 = spark.createDataFrame(data=dat3, schema=col3)
df3 = df3.withColumn("event_start", to_timestamp(col("event_start")))


# make temp hive view
df3.createOrReplaceTempView("event_data3")

# show schema
df3.printSchema()

The image below shows the date that has a year of '2' is converted to a null value since it is not valid. This malformed data will blow up the timestamp example above.

In short, know your incoming data. Profile the data for bad values. Then determine which method is best to load the data. Always remember, supplying a schema results in a faster load time for some types of files.

How can I let PySpark recognize a column as a datetime type?

2 Answers