Spark PythonからPandasへのタイムスタンプ往復

Question

Spark PythonからPandasにタイムスタンプデータを往復変換する方法はありますか？私はSparkのHiveテーブルからデータを読み込み、Pandasでいくつかの計算を行い、結果をHiveに書き戻したいと思います。最後の部分だけが失敗し、PandasタイムスタンプをSpark DataFrameタイムスタンプに変換し直します。

import datetime 
import pandas as pd 

dates = [ 
    ('today', '2017-03-03 11:30:00') 
    , ('tomorrow', '2017-03-04 08:00:00') 
    , ('next Thursday', '2017-03-09 20:00:00') 
] 
string_date_rdd = sc.parallelize(dates) 
timestamp_date_rdd = string_date_rdd.map(lambda t: (t[0], datetime.datetime.strptime(t[1], "%Y-%m-%d %H:%M:%S'))) 
timestamp_df = sqlContext.createDataFrame(timestamp_date_rdd, ['Day', 'Date']) 
timestamp_pandas_df = timestamp_df.toPandas() 
roundtrip_df = sqlContext.createDataFrame(timestamp_pandas_df) 
roundtrip_df.printSchema() 
roundtrip_df.show() 

root 
|-- Day: string (nullable = true) 
|-- Date: long (nullable = true) 

+-------------+-------------------+ 
|   Day|    Date| 
+-------------+-------------------+ 
|  today|1488540600000000000| 
|  tomorrow|1488614400000000000| 
|next Thursday|1489089600000000000| 
+-------------+-------------------+ 

この時点で、往復のSpark DataFrameには、日付型の列がデータ型longとしてあります。 Pysparkでは、これはdatetime.datetime.fromtimestamp（148908960000000000/1000000000）のようにdatetimeオブジェクトに簡単に変換できますが、時刻は数時間ずれています。 Spark DataFrameのデータ型を変換するにはどうすればよいですか？ Pythonの3.4.5

、1.6.0

おかげスパーク、Pythonの日時にdatetime64列を変換 ジョン

Answer 1

は私のために作品をオブジェクト。

from pandas import Series 

def convert_to_python_datetime(df): 
    df_copy = df.copy() 
    for column_name, column in df_copy.iteritems(): 
     if column.dtype.kind == 'M': 
      df_copy[column_name] = Series(column.dt.to_pydatetime(), dtype=object) 
    return df_copy 


tmp = convert_to_python_datetime(timestamp_pandas_df) 
roundtrip_df = sqlContext.createDataFrame(tmp) 
roundtrip_df.printSchema() 
roundtrip_df.show() 

出力：

from pyspark.sql.types import TimestampType 
extra_column_df = roundtrip_df.select(roundtrip_df.Day, roundtrip_df.Date).withColumn('new_date', roundtrip_df.Date/1000000000) 
roundtrip_timestamp_df = extra_column_df.select(extra_column_df.Day, extra_column_df.new_date.cast(TimestampType()).alias('Date') 

出力：：

root 
|-- Day: string (nullable = true) 
|-- Date: timestamp (nullable = true) 

+-------------+--------------------+ 
|   Day|    Date| 
+-------------+--------------------+ 
|  today|2017-03-03 11:30:...| 
|  tomorrow|2017-03-04 08:00:...| 
|next Thursday|2017-03-09 20:00:...| 
+-------------+--------------------+ 

Answer 2

ここで私が見つけた一つの解決策だ追加のバグや機能として

root 
|-- Day: string (nullable = true) 
|-- Date: timestamp (nullable = true) 

+-------------+--------------------+ 
|  Day |    Date| 
+-------------+--------------------+ 
|  today|2017-03-03 11:30:...| 
|  tomorrow|2017-03-04 08:00:...| 
|next Thursday|2017-03-09 20:00:...| 
+-------------+--------------------+ 

が、これはすべて変換するようですDSTを含むUTCの日付エッセンス