python - PySpark:将 "string-integer"列转换为 IntegerType

Question

我有一个以 datetime.datetime 对象作为其内容的列。我正在尝试使用 pyspark.sql.Window 功能，该功能需要数字类型，而不是日期时间或字符串。所以我的计划是将 datetime.datetime 对象转换为 UNIX 时间戳:

设置:

>>> import datetime; df = sqlContext.createDataFrame(
... [(datetime.datetime(2018, 1, 17, 19, 0, 15),),
... (datetime.datetime(2018, 1, 17, 19, 0, 16),)], ['dt'])
>>> df
DataFrame[dt: timestamp]
>>> df.dtypes
[('dt', 'timestamp')]
>>> df.show(5, False)
+---------------------+
|dt                   |
+---------------------+
|2018-01-17 19:00:15.0|
|2018-01-17 19:00:16.0|
+---------------------+

定义一个函数来访问datetime.datetime对象的timestamp函数:

def dt_to_timestamp():
    def _dt_to_timestamp(dt):
        return int(dt.timestamp() * 1000)
    return func.udf(_dt_to_timestamp)

应用该函数:

>>> df = df.withColumn('dt_ts', dt_to_timestamp()(func.col('dt')))
>>> df.show(5, False)
+---------------------+-------------+
|dt                   |dt_ts        |
+---------------------+-------------+
|2018-01-17 19:00:15.0|1516237215000|
|2018-01-17 19:00:16.0|1516237216000|
+---------------------+-------------+

>>> df.dtypes
[('dt', 'timestamp'), ('dt_ts', 'string')]

我不确定当内部 _dt_to_timestamp 函数返回 int 时，为什么此列默认为 string，但让我们尝试对这些进行强制转换“字符串整数”到IntegerTypes:

>>> df = df.withColumn('dt_ts', func.col('dt_ts').cast(IntegerType()))
>>> df.show(5, False)
+---------------------+-----+
|dt                   |dt_ts|
+---------------------+-----+
|2018-01-17 19:00:15.0|null |
|2018-01-17 19:00:16.0|null |
+---------------------+-----+

>>> df.dtypes
[('dt', 'timestamp'), ('dt_ts', 'int')]

这似乎只是 IntegerType 强制转换的问题。对于 DoubleType，转换有效，但我更喜欢整数...

>>> df = df.withColumn('dt_ts', dt_to_timestamp()(func.col('dt')))
>>> df = df.withColumn('dt_ts', func.col('dt_ts').cast(DoubleType()))
>>> df.show(5, False)
+---------------------+--------------+
|dt                   |dt_ts         |
+---------------------+--------------+
|2018-01-17 19:00:15.0|1.516237215E12|
|2018-01-17 19:00:16.0|1.516237216E12|
+---------------------+--------------+

Answer 1

这是因为 IntegerType 无法存储与您尝试转换一样大的数字。请改用 bigint/long 类型:

>>> df = df.withColumn('dt_ts', dt_to_timestamp()(func.col('dt')))
>>> df.show()
+--------------------+-------------+
|                  dt|        dt_ts|
+--------------------+-------------+
|2018-01-17 19:00:...|1516237215000|
|2018-01-17 19:00:...|1516237216000|
+--------------------+-------------+

>>> df = df.withColumn('dt_ts', func.col('dt_ts').cast('long'))
>>> df.show()
+--------------------+-------------+
|                  dt|        dt_ts|
+--------------------+-------------+
|2018-01-17 19:00:...|1516237215000|
|2018-01-17 19:00:...|1516237216000|
+--------------------+-------------+

>>> df.dtypes
[('dt', 'timestamp'), ('dt_ts', 'bigint')]