错误：从列表创建 Spark 数据帧时 TimestampType 无法接受对象

我正在尝试从以下列表创建一个数据框：

data = [(1,'abc','2020-08-20 10:00:00', 'I'),
(1,'abc','2020-08-20 10:01:00', 'U'),
(1,'abc','2020-08-21 10:02:00', 'U'),
(2,'pqr','2020-08-20 10:00:00', 'I'),
(2,'pqr','2020-08-20 10:01:00', 'U'),
(2,'pqr','2020-08-21 10:02:00', 'D'),
(3,'rst','2020-08-20 10:00:00', 'I'),
(3,'rst','2020-08-20 10:01:00', 'U'),
(3,'rst','2020-08-21 10:02:00', 'U')]

我正在运行以下代码来创建一个数据框：

from pyspark.sql.types import *
mySchema = StructType([StructField("key", IntegerType()),
                      StructField("name", StringType()),
                      StructField("ts", TimestampType()),
                      StructField("cdc_flag", StringType())])

df_raw = spark.createDataFrame(data, mySchema)

我收到以下错误：

TypeError: field ts: TimestampType can not accept object '2020-08-20 10:00:00' in type <class 'str'>

我也尝试将数据类型更改为 DateType 。但遇到同样的错误。

请注意，我试图了解这种实现模式的方式是否可行。 我想我可以使用 withColumn 并强制转换此 ts 列并删除原始列来处理此问题。

该错误是合理的，因为TimestampType https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.types.TimestampType.html期望一个Timestamp类型而不是str。这可以通过使用得出java.sql.Timestamp在斯卡拉和datetime在Python中。

你只需要定义你的data like:

from datetime import datetime

data = [(1,'abc',datetime.strptime('2020-08-20 10:00:00', '%Y-%m-%d %H:%M:%S'), 'I'),
(1,'abc',datetime.strptime('2020-08-20 10:01:00', '%Y-%m-%d %H:%M:%S'), 'U'),
(1,'abc',datetime.strptime('2020-08-21 10:02:00', '%Y-%m-%d %H:%M:%S'), 'U'),
(2,'pqr',datetime.strptime('2020-08-20 10:00:00', '%Y-%m-%d %H:%M:%S'), 'I'),
(2,'pqr',datetime.strptime('2020-08-20 10:01:00', '%Y-%m-%d %H:%M:%S'), 'U'),
(2,'pqr',datetime.strptime('2020-08-21 10:02:00', '%Y-%m-%d %H:%M:%S'), 'D'),
(3,'rst',datetime.strptime('2020-08-20 10:00:00', '%Y-%m-%d %H:%M:%S'), 'I'),
(3,'rst',datetime.strptime('2020-08-20 10:01:00', '%Y-%m-%d %H:%M:%S'), 'U'),
(3,'rst',datetime.strptime('2020-08-21 10:02:00', '%Y-%m-%d %H:%M:%S'), 'U')]


spark.createDataFrame(data, mySchema).show()
#+---+----+-------------------+--------+
#|key|name|                 ts|cdc_flag|
#+---+----+-------------------+--------+
#|  1| abc|2020-08-20 10:00:00|       I|
#|  1| abc|2020-08-20 10:01:00|       U|
#|  1| abc|2020-08-21 10:02:00|       U|
#|  2| pqr|2020-08-20 10:00:00|       I|
#|  2| pqr|2020-08-20 10:01:00|       U|
#|  2| pqr|2020-08-21 10:02:00|       D|
#|  3| rst|2020-08-20 10:00:00|       I|
#|  3| rst|2020-08-20 10:01:00|       U|
#|  3| rst|2020-08-21 10:02:00|       U|
#+---+----+-------------------+--------+


spark.createDataFrame(data, mySchema).printSchema()
#root
# |-- key: integer (nullable = true)
# |-- name: string (nullable = true)
# |-- ts: timestamp (nullable = true)
# |-- cdc_flag: string (nullable = true)