我有以下问题。我有一个数据集来跟踪状态的变化。

id  valid  eventdate
 1  False 2020-05-01
 1   True 2020-05-06
 2   True 2020-05-04
 2  False 2020-05-07
 2   True 2020-05-09
 3  False 2020-05-11

目标：

SELECT valid FROM table WHERE id = 1 AND eventdate = "2020-05-05"

我需要知道在任何给定的日期（在开始和今天）的状态是一个给定的一天。例如，例如 id 1有效期仍然有效 False 五月五日。
在Pandas中，我有一个这样的解决方案，我使用 pivot 以及 ffill 用于填充空值。我使用一个melt将它重新制作成一个三列的Dataframe。

from datetime import datetime
import pandas as pd
test_data = [
  [1,"False","2020-05-01"],
  [1,"True","2020-05-06"],
  [2,"True","2020-05-04"],
  [2,"False","2020-05-07"],
  [2,"True","2020-05-09"],
  [3,"False","2020-05-11"]
]
# Create inputframe
df = pd.DataFrame(test_data, columns=['id', 'valid', 'eventdate'])
df['id'] = df['id'].astype(str)
df['valid'] = df['valid'] == "True"
df['eventdate'] = pd.to_datetime(df['eventdate'])
print(df.head(6))
# id  valid  eventdate
# 0  1  False 2020-05-01
# 1  1   True 2020-05-06
# 2  2   True 2020-05-04
# 3  2  False 2020-05-07
# 4  2   True 2020-05-09
# 5  3  False 2020-05-11
# Create full time range as frame
timeframe = pd.date_range(start=min(df['eventdate']),
                          end=datetime.now().date()).to_frame().reset_index(drop=True).rename(columns={0: 'eventdate'})
print(timeframe.head())
# eventdate
# 0 2020-05-01
# 1 2020-05-02
# 2 2020-05-03
# 3 2020-05-04
# 4 2020-05-05
# Merge timeframe into original frame
df = df.merge(timeframe,
              left_on='eventdate',
              right_on='eventdate',
              how='right')
print(df.sort_values('eventdate').head())
# id  valid  eventdate
# 0    1  False 2020-05-01
# 6  NaN    NaN 2020-05-02
# 7  NaN    NaN 2020-05-03
# 2    2   True 2020-05-04
# 8  NaN    NaN 2020-05-05
# 1. Pivot to get dates on rows and ids as columns
# 2. Forward fill values per id
# 3. Fill remaining NaNs with False
df = df.pivot(index='eventdate',
              columns='id',
              values='valid')\
       .fillna(method='ffill')\
       .fillna(False)
print(df.head())
# id            NaN      1      2      3
# eventdate
# 2020-05-01  False  False  False  False
# 2020-05-02  False  False  False  False
# 2020-05-03  False  False  False  False
# 2020-05-04  False  False   True  False
# 2020-05-05  False  False   True  False
# Drop NaN column and reset the index
df = df.loc[:, df.columns.notnull()].reset_index()
# Melt the columns back
out = pd.melt(df,
              id_vars='eventdate',
              value_name='valid')
print(out.head(10))
# eventdate id  valid
# 0 2020-05-01  1  False
# 1 2020-05-02  1  False
# 2 2020-05-03  1  False
# 3 2020-05-04  1  False
# 4 2020-05-05  1  False
# 5 2020-05-06  1   True
# 6 2020-05-07  1   True
# 7 2020-05-08  1   True
# 8 2020-05-09  1   True
# 9 2020-05-10  1   True

我正在努力实现同样的Spark，但向前填补不存在。我知道如何达到最新的地位 id :

w = Window().partitionBy("id").orderBy(F.col("eventdate").desc())
df.withColumn("rn", F.row_number().over(w)) \
  .where(F.col("rn") == 1) \
  .selectExpr("id", "valid", "eventdate AS last_change") \
  .dropna() \
  .show()

旋转可通过以下方式完成：

df\
.select(["id", "valid", "eventdate"])\
.groupBy(["eventdate"])\
.pivot("id")\
.agg(F.min("valid"))\
.drop('null')\
.sort('eventdate')\
.show()

为了进行正向填充，我将数据集限制为一个 id :

import sys
from datetime import datetime
import pyspark.sql.functions as F
from pyspark.sql import Window
test_data = [
  [1,"False","2020-05-01"],
  [1,"True","2020-05-06"],
  [2,"True","2020-05-04"],
  [2,"False","2020-05-07"],
  [2,"True","2020-05-09"],
  [3,"False","2020-05-11"]
]
# Create dataframe
df = sc\
  .parallelize(test_data)\
  .toDF(("id", "valid", "eventdate"))\
  .withColumn("eventdate", F.to_date(F.to_timestamp("eventdate")))\
  .withColumn("valid", F.when(F.col("valid") == "True", 1).otherwise(0))
df.createOrReplaceTempView("df")
# Create event frame
event_dates = spark.sql("SELECT sequence(min(eventdate), CURRENT_DATE(), interval 1 day) as eventdate FROM df")\
                   .withColumn("eventdate",
                               F.explode(F.col("eventdate")))
# Join dates and data
df = df.join(event_dates, on='eventdate', how='right')
df2 = df.where(df.id == 1)\
  .join(event_dates, on='eventdate', how='right')\
  .withColumn('id', F.lit(1))
# df2.sort('eventdate').show()
# +----------+---+-----+
# | eventdate| id|valid|
# +----------+---+-----+
# |2020-05-01|  1|    0|
# |2020-05-02|  1| null|
# |2020-05-03|  1| null|
# |2020-05-04|  1| null|
# |2020-05-05|  1| null|
# |2020-05-06|  1|    1|
# |2020-05-07|  1| null|
# |2020-05-08|  1| null|
# |2020-05-09|  1| null|
# |2020-05-10|  1| null|
# |2020-05-11|  1| null|
# |2020-05-12|  1| null|
# |2020-05-13|  1| null|
# +----------+---+-----+
# Forward fill
window = Window.partitionBy('id')\
               .orderBy('eventdate')\
               .rowsBetween(-sys.maxsize, 0)
# Set filter
read_last = F.last(df2['valid'], ignorenulls=True).over(window)
df2.withColumn("ffill", read_last).show()
# +----------+---+-----+-----+
# | eventdate| id|valid|ffill|
# +----------+---+-----+-----+
# |2020-05-01|  1|    0|    0|
# |2020-05-02|  1| null|    0|
# |2020-05-03|  1| null|    0|
# |2020-05-04|  1| null|    0|
# |2020-05-05|  1| null|    0|
# |2020-05-06|  1|    1|    1|
# |2020-05-07|  1| null|    1|
# |2020-05-08|  1| null|    1|
# |2020-05-09|  1| null|    1|
# |2020-05-10|  1| null|    1|
# |2020-05-11|  1| null|    1|
# |2020-05-12|  1| null|    1|
# |2020-05-13|  1| null|    1|
# +----------+---+-----+-----+

我认为第一件事是这个回答问题的方法是否正确。做 pivot 将创建一个包含少数列的长表，同时存储大量冗余数据。spark不是解决问题的合适工具，或者更好，问题本身不适合使用spark。我知道理想情况下，您需要使用并行处理，也许还需要广播 timeframe 并计算每个节点的正向填充 id 每个节点？
是否最好使用一些不同的方法，例如，存储 enddate 在查询时使用以下内容：

id  valid  eventdate enddate
 1  False 2020-05-01 2020-05-06
 1   True 2020-05-06 2999-12-31
 2   True 2020-05-04 2020-05-07
 2  False 2020-05-07 2020-05-08
 2   True 2020-05-09 2999-12-31
 3  False 2020-05-11 2999-12-31

以及

SELECT valid FROM table WHERE id = 1 AND "2020-05-05" between eventdate and enddate

请让我知道spark方法是否有用，对于这样一个稀疏的数据集，在任何给定的日历状态下查找状态的最佳方法是什么？
谢谢您。