从 pandas 到 pyspark - 将带有开始和结束日期的数据框转换为每日数据？答案

【问题标题】：Going from pandas to pyspark - Convert dataframe with start and end date to daily data?从 pandas 到 pyspark - 将带有开始和结束日期的数据框转换为每日数据？
【发布时间】：2020-03-04 16:57:29
【问题描述】：

我有一个使用 Pandas 的 ETL 脚本，为了使其更具可扩展性，我尝试使用 Pyspark 重新创建它。到目前为止一切顺利，但在对每日数据集的特定转换方面遇到了问题。我每个 ID 有一条记录，包括开始日期和结束日期

id  age state   start_date  end_date
123 18  CA     2/17/2019    5/4/2019
223 24  AZ     1/17/2019    3/4/2019

我想为开始日和结束日之间的每一天创建一条记录，以便将每日活动数据加入其中。目标输出看起来像这样

id  age state   start_date
123 18  CA      2/17/2019
123 18  CA      2/18/2019
123 18  CA      2/19/2019
123 18  CA      2/20/2019
123 18  CA      2/21/2019
            …
123 18  CA      5/2/2019
123 18  CA      5/3/2019
123 18  CA      5/4/2019

当然，对数据集中的所有 id 及其各自的开始日期执行此操作。我可以使用以下方法在 Pandas 中做到这一点

melt = df.melt(id_vars=['id', 'age', 'state'], value_name='date').drop('variable', axis=1)
melt['date'] = pd.to_datetime(melt['date'])

melt = melt.groupby('id').apply(lambda x: x.set_index('date').resample('d').first())\
           .ffill()\
           .reset_index(level=1)\
           .reset_index(drop=True)

但我对 Pyspark 相当陌生（并且在 Pandas 中一直在努力解决这个问题）所以我被困在这里。非常感谢任何帮助 - 谢谢！

【问题讨论】：

这能回答你的问题吗？ PySpark: how to resample frequencies
看起来绝对是个不错的起点。谢谢！
所以这越来越近了，但我不清楚如何跨多个 id 做到这一点？基本上，在示例中，他创建了一个具有所需间隔的时间序列，然后将来自单个变量的观察结果连接回它。但在我的情况下，每个 ID 都应该有每日间隔的范围，并且这些间隔对于每个 ID 都是不同的。

标签： pandas apache-spark pyspark reshape melt

【解决方案1】：

在this post 中找到了解决方案。我的解决方案的关键是explode 函数，它可以满足我的需要。

解决我的具体例子的代码是

def date_range(t1, t2, step=60*60*24):
    return [t1 + step*x for x in range(int((t2-t1)/step)+1)]

date_range_udf = udf(date_range, ArrayType(LongType()))

df = dataF.select("id",
expr("stack(2, 'start_date', start_date, 'end_date', end_date) as (class_date,date)"))

df_base = \
    df.groupBy('id')\
        .agg(min('date').cast('integer').alias('date_min'), max('date').cast('integer')\
    .alias('date_max'))\
        .withColumn("date", explode(date_range_udf("date_min", "date_max")))\
        .drop('date_min', 'date_max')\
        .withColumn("date", from_unixtime("date"))

它给出了以下输出（我可以用它来连接任何额外的数据）

【讨论】：