pyspark 在对该列使用 groupby 之前更改该列的值答案

【问题标题】：pyspark Change the value of a column before using groupby on that columnpyspark 在对该列使用 groupby 之前更改该列的值
【发布时间】：2017-03-01 09:14:55
【问题描述】：

我有这个 json 数据，我想每小时汇总一次“时间戳”列，同时汇总“b”和“a”列中的数据。

{"a":1 , "b":1, "timestamp":"2017-01-26T01:14:55.719214Z"}
{"a":1 , "b":1,"timestamp":"2017-01-26T01:14:55.719214Z"}
{"a":1 , "b":1,"timestamp":"2017-01-26T02:14:55.719214Z"}
{"a":1 , "b":1,"timestamp":"2017-01-26T03:14:55.719214Z"}

这是我想要的最终输出

{"a":2 , "b":2, "timestamp":"2017-01-26T01:00:00"}
{"a":1 , "b":1,"timestamp":"2017-01-26T02:00:00"}
{"a":1 , "b":1,"timestamp":"2017-01-26T03:00:00"}

这是我到目前为止写的内容

df = spark.read.json(inputfile)
df2 = df.groupby("timestamp").agg(f.sum(df["a"],f.sum(df["b"])

但是在使用 groupby 函数之前，我应该如何更改 'timestamp' 列的值？提前致谢！

【问题讨论】：

这个answer 可能会有所帮助。它显示了如何对解析的时间戳对象进行舍入。

标签： apache-spark pyspark apache-spark-sql spark-streaming pyspark-sql

【解决方案1】：

from pyspark.sql import functions as f   

df = spark.read.load(path='file:///home/zht/PycharmProjects/test/disk_file', format='json')
df = df.withColumn('ts', f.to_utc_timestamp(df['timestamp'], 'EST'))
win = f.window(df['ts'], windowDuration='1 hour')
df = df.groupBy(win).agg(f.sum(df['a']).alias('sumA'), f.sum(df['b']).alias('sumB'))
res = df.select(df['window']['start'].alias('start_time'), df['sumA'], df['sumB'])
res.show(truncate=False)

# output:
+---------------------+----+----+                                               
|start_time           |sumA|sumB|
+---------------------+----+----+
|2017-01-26 15:00:00.0|1   |1   |
|2017-01-26 16:00:00.0|1   |1   |
|2017-01-26 14:00:00.0|2   |2   |
+---------------------+----+----+

f.window 更加灵活

【讨论】：

感谢您的回答，实际上我只需要时间戳列中的 '2017-01-26 15:00:00.0' 而不是 '[2017-01-26 15:00:00.0,2017-01 -26 16:00:00.0]'。你知道我怎样才能得到这个吗？

【解决方案2】：

我想这是做到这一点的一种方法

df2 = df.withColumn("r_timestamp",df["r_timestamp"].substr(0,12)).groupby("timestamp").agg(f.sum(df["a"],f.sum(df["b"])

有没有更好的解决方案来获取所需格式的时间戳？

【讨论】：