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

我有一个数据框，其结构大致如下:

+-------------------------+-------------------------+--------+ 
| 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 行多得多，是否有可能数据正在异步处理，因此生成的数据帧的顺序与输入的顺序不同？

请您参考如下方法:

对于 udf，我不太确定它为什么不起作用。将 Python 函数转换为 UDF 时，可能是 float manipulation 问题。看看下面如何使用整数输出。或者，您可以使用名为 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) 

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

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]。