Преобразование метки времени в UTC в Spark Scala - PullRequest
Новая
       16

Преобразование метки времени в UTC в Spark Scala

0 голосов
/ 04 мая 2018

Моя среда - Spark 2.1, Scala

Это может быть просто, но я ломаю голову.

Мой Dataframe, myDF похож на ниже - 

+--------------------+----------------+  
|     orign_timestamp | origin_timezone|  
+--------------------+----------------+  
|2018-05-03T14:56:...|America/St_Johns|  
|2018-05-03T14:56:...| America/Toronto|  
|2018-05-03T14:56:...| America/Toronto|    
|2018-05-03T14:56:...| America/Toronto|  
|2018-05-03T14:56:...| America/Halifax|  
|2018-05-03T14:56:...| America/Toronto|  
|2018-05-03T14:56:...| America/Toronto|  
+--------------------+----------------+

Мне нужно преобразовать orign_timestamp в UTC и добавить новый столбец в DF. Код ниже работает нормально. 

myDF.withColumn("time_utc", to_utc_timestamp(from_unixtime(unix_timestamp(col("orign_timestamp"), "yyyy-MM-dd'T'HH:mm:ss")),("America/Montreal"))).show

Проблема в том, что я установил часовой пояс на "Америка / Монреаль". Мне нужно передать столбец timeZone "orign_timeone" столбец. Я пытался 

myDF.withColumn("time_utc", to_utc_timestamp(from_unixtime(unix_timestamp(col("orign_timestamp"), "yyyy-MM-dd'T'HH:mm:ss")), col("orign_timezone".toString.trim))).show

got Error:
<console>:34: error: type mismatch;
 found   : org.apache.spark.sql.Column
 required: String

Я попробовал приведенный ниже код, но не через исключение, но новый столбец имел то же время, что и origin_time. 

myDF.withColumn("origin_timestamp", to_utc_timestamp(from_unixtime(unix_timestamp(col("orign_timestamp"), "yyyy-MM-dd'T'HH:mm:ss")), col("rign_timezone").toString)).show

1 Ответ

0 голосов
/ 04 мая 2018

Всякий раз, когда вы сталкиваетесь с такой проблемой, вы можете использовать expr

import org.apache.spark.sql.functions._

val df = Seq(
  ("2018-05-03T14:56:00", "America/St_Johns"), 
  ("2018-05-03T14:56:00", "America/Toronto"), 
  ("2018-05-03T14:56:00", "America/Halifax")
).toDF("origin_timestamp", "origin_timezone")

df.withColumn("time_utc",
  expr("to_utc_timestamp(origin_timestamp, origin_timezone)")
).show

// +-------------------+----------------+-------------------+
// |   origin_timestamp| origin_timezone|           time_utc|
// +-------------------+----------------+-------------------+
// |2018-05-03T14:56:00|America/St_Johns|2018-05-03 17:26:00|
// |2018-05-03T14:56:00| America/Toronto|2018-05-03 18:56:00|
// |2018-05-03T14:56:00| America/Halifax|2018-05-03 17:56:00|
// +-------------------+----------------+-------------------+

или selectExpr: 

df.selectExpr(
  "*", "to_utc_timestamp(origin_timestamp, origin_timezone) as time_utc"
).show

// +-------------------+----------------+-------------------+
// |   origin_timestamp| origin_timezone|           time_utc|
// +-------------------+----------------+-------------------+
// |2018-05-03T14:56:00|America/St_Johns|2018-05-03 17:26:00|
// |2018-05-03T14:56:00| America/Toronto|2018-05-03 18:56:00|
// |2018-05-03T14:56:00| America/Halifax|2018-05-03 17:56:00|
// +-------------------+----------------+-------------------+
...