pyspark - 错误 : TimestampType can not accept object while creating a Spark dataframe from a list

Question

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

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 列并删除原始列来处理此问题。

Answer 1

该错误是合理的，因为 TimestampType 期待 Timestamp类型而不是 str .这可以通过使用 java.sql.Timestamp 推导出来在 Scala 和 datetime 中在 Python 中。
您只需要定义您的 data喜欢:

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)