apache-spark - pyspark date_format() 和 hour() 将时间戳转换为本地时间

Question

编辑:我使用的是 pyspark 2.0.2，无法使用更高版本。

我有一些源数据的时间戳字段偏移量为零，我只是想从该字段中提取日期和小时。但是，spark 在检索日期和小时之前将此时间戳转换为本地时间(在我的例子中是美国东部时间)。使用 UDF 从时间戳字段中剥离 T 和 Z 并应用上述相同的函数是可行的，但这似乎是一种愚蠢的方式来满足我的需要。有什么想法吗？

from pyspark.sql import SparkSession
from pyspark.sql.functions import date_format, hour

spark = (
    SparkSession
    .builder
    .appName('junk')
    .getOrCreate()
    )
spark.sparkContext.setLogLevel('ERROR')

df = spark.createDataFrame(
    [(1, '2018-04-20T00:56:30.562Z'),
     (2, '2018-04-20T03:56:30.562Z'),
     (3, '2018-04-20T05:56:30.562Z')],
    ['id', 'ts']
    )
df = (
    df
    .withColumn(
        'event_dt',
        date_format(df.ts.cast('timestamp'), 'yyyy-MM-dd').cast('date')
        )
    .withColumn('event_hr', hour(df.ts))
    )
print(df.head(5))

输出如下:

[
  Row(id=1, ts='2018-04-20T00:56:30.562Z', event_dt=datetime.date(2018, 4, 19), event_hr=20),
  Row(id=2, ts='2018-04-20T03:56:30.562Z', event_dt=datetime.date(2018, 4, 19), event_hr=23),
  Row(id=3, ts='2018-04-20T05:56:30.562Z', event_dt=datetime.date(2018, 4, 20), event_hr=1)
]

以下解决方法有效，但如果可能的话，我正在寻找更直接的方法:

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

stripTz = udf(lambda x: x.replace('T', ' ').replace('Z', ''), StringType())
df = (
    df
    .withColumn('newts', stripTz(df.ts))
    )
df = (
    df
    .withColumn(
        'event_dt',
        date_format(df.newts.cast('timestamp'), 'yyyy-MM-dd').cast('date')
        )
    .withColumn('event_hr', hour(df.newts))
    .drop('newts')
    )

print(df.head(5))

新的输出如下和需要的:

[
  Row(id=1, ts='2018-04-20T00:56:30.562Z', event_dt=datetime.date(2018, 4, 20), event_hr=0),
  Row(id=2, ts='2018-04-20T03:56:30.562Z', event_dt=datetime.date(2018, 4, 20), event_hr=3),
  Row(id=3, ts='2018-04-20T05:56:30.562Z', event_dt=datetime.date(2018, 4, 20), event_hr=5)
]

Answer 1

您使用的是什么版本的 Spark？在 2.2+ 中，只需将您的 sparksession 的时区设置为:

spark.conf.set("spark.sql.session.timeZone", "GMT")

或者，

df.select("id", "ts", pyspark.sql.functions.to_timestamp("ts").alias("timestamp"))

然后在提取日期/小时之前将时区更改为任何时间