pyspark 子字符串和聚合答案

【问题标题】：pyspark substring and aggregationpyspark 子字符串和聚合
【发布时间】：2017-01-07 21:22:26
【问题描述】：

我是 Spark 的新手，我有一个包含此类数据的 csv 文件：

date,            accidents, injured
2015/20/03 18:00    15,          5
2015/20/03 18:30    25,          4
2015/20/03 21:10    14,          7
2015/20/02 21:00    15,          6

我想按事件发生的特定时间汇总这些数据。我的想法是在没有分钟的情况下将日期子串为“年/月/日 hh”，这样我就可以将其设为密钥。我想给出每小时平均事故和受伤人数。也许 pyspark 有另一种更智能的方式？

谢谢大家！

【问题讨论】：

标签： substring pyspark aggregate

【解决方案1】：

嗯，我猜这取决于你之后要做什么。

最简单的方法是按照您的建议进行操作：将日期字符串子串化，然后聚合：

data = [('2015/20/03 18:00', 15, 5), 
    ('2015/20/03 18:30', 25, 4),
    ('2015/20/03 21:10', 14, 7),
    ('2015/20/02 21:00', 15, 6)]
df = spark.createDataFrame(data, ['date', 'accidents', 'injured'])

df.withColumn('date_hr',
              df['date'].substr(1, 13)
     ).groupby('date_hr')\
      .agg({'accidents': 'avg', 'injured': 'avg'})\
      .show()

但是，如果您以后想要进行更多计算，可以将数据解析为 TimestampType()，然后从中提取日期和时间。

import pyspark.sql.types as typ
from pyspark.sql.functions import col, udf
from datetime import datetime

parseString =  udf(lambda x: datetime.strptime(x, '%Y/%d/%m %H:%M'),   typ.TimestampType())
getDate =  udf(lambda x: x.date(), typ.DateType())
getHour = udf(lambda x: int(x.hour), typ.IntegerType())

df.withColumn('date_parsed', parseString(col('date'))) \
    .withColumn('date_only', getDate(col('date_parsed'))) \
    .withColumn('hour', getHour(col('date_parsed'))) \
    .groupby('date_only', 'hour') \
    .agg({'accidents': 'avg', 'injured': 'avg'})\
    .show()

【讨论】：

在使用 y[0][:13] 很容易地进行映射时设法创建子字符串。相信您的解决方案看起来更优雅。谢谢！还有一个问题，如果我有另一个包含其他数据的文件，假设来自不同的年份，我将如何获得事故和伤害的平均值？将所有内容放在一个文件中然后运行计算？
我要么读取该文件并仅对该数据进行聚合，要么，如果您需要一次性获得结果（假设您使用 Spark 2.0），您可以 @987654325 @ 两个（或更多）DataFrames 一起。 spark.apache.org/docs/latest/api/python/…