PySpark：在数据帧中将时间戳转换为整数的不一致

Question

我有一个大致结构如下的数据框：

+-------------------------+-------------------------+--------+
| timestamp               | adj_timestamp           | values |
+-------------------------+-------------------------+--------+
| 2017-05-31 15:30:48.000 | 2017-05-31 11:30:00.000 | 0      |
+-------------------------+-------------------------+--------+
| 2017-05-31 15:31:45.000 | 2017-05-31 11:30:00.000 | 0      |
+-------------------------+-------------------------+--------+
| 2017-05-31 15:32:49.000 | 2017-05-31 11:30:00.000 | 0      |
...

我正在尝试使用time 包将转换函数应用于两个时间列，以将它们转换为整数表示。我的用户定义函数以及它如何应用于上面的数据框：

def timeConverter(timestamp):
    time_tuple = time.strptime(timestamp, "%Y-%m-%d %H:%M:%S.000")
    timevalue = time.mktime(time_tuple)
    return timevalue

def convertDateColumn(Data):
    timeUDF = udf(timeConverter,FloatType())
    finalData = Data.withColumn('adj_timestamp', timeUDF('adj_timestamp'))

    return finalData

例如，adj_timestamp 列中的第一个条目变为： 1496244608

通过datetime.fromtimestamp 将其转换回：2017-05-31 15:30:08

这与我一开始的价值不同...很好奇发生了什么！

编辑：由于我的行数远多于所示的 3 行，是否有可能异步处理数据，因此生成的数据帧与输入的顺序不同？

Answer 1

对于udf，我不太确定为什么它不起作用。将 Python 函数转换为 UDF 时可能是浮点操作问题。看看下面如何使用整数输出。或者，您可以使用名为 unix_timestamp 的 Spark 函数来解决问题，该函数允许您转换时间戳。我在下面举一个例子。希望对您有所帮助。

在这里，我根据您展示的示例创建 Spark 数据框，

import pandas as pd

df = pd.DataFrame([
    ['2017-05-31 15:30:48.000', '2017-05-31 11:30:00.000', 0], 
    ['2017-05-31 15:31:45.000', '2017-05-31 11:30:00.000', 0],
    ['2017-05-31 15:32:49.000', '2017-05-31 11:30:00.000', 0]], 
    columns=['timestamp', 'adj_timestamp', 'values'])
df = spark.createDataFrame(df)

使用Spark函数求解

将fn.unix_timestamp应用于timestamp列

import pyspark.sql.functions as fn
from pyspark.sql.types import *
df.select(fn.unix_timestamp(fn.col('timestamp'), format='yyyy-MM-dd HH:mm:ss.000').alias('unix_timestamp')).show()

对于第一列，输出如下所示

+--------------+
|unix_timestamp|
+--------------+
|    1496259048|
|    1496259105|
|    1496259169|
+--------------+

您可以使用 datetime 库将其放回时间戳：

import datetime
datetime.datetime.fromtimestamp(1496259048) # output as datetime(2017, 5, 31, 15, 30, 48)

通过转换为整数而不是浮点数来解决

import datetime
import time

def timeConverter(timestamp):
    time_tuple = datetime.datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S.000").timetuple()
    timevalue = int(time.mktime(time_tuple)) # convert to int here
    return timevalue

time_udf = fn.udf(timeConverter, IntegerType()) # output interger

df.select(time_udf(fn.col('timestamp'))) 

在这里，我们将获得与使用unix_timestamp 相同的时间戳[1496259048, 1496259105, 1496259169]。