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. Посмотрите, как использование выходного интергера работает ниже. Кроме того, вы можете решить, используя функцию Spark под названием unix_timestamp это позволяет вам конвертировать метку времени. Я приведу пример ниже. Надеюсь, это поможет немного.

Здесь я создаю Spark DataFrame из примеров, которые вы показываете,

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'))) 

Здесь мы получим ту же метку времени [1496259048, 1496259105, 1496259169] как используя unix_timestamp,