Pyspark如何过滤RDD Map Function中的数据框？答案

【问题标题】：Pyspark how to filter a dataframe inside RDD Map Function?Pyspark如何过滤RDD Map Function中的数据框？
【发布时间】：2021-10-15 19:10:15
【问题描述】：

我正在尝试在地图回调函数中查询数据框，并使用基于多行和多列的计算创建一个新列。

DF 看起来像这样

datum	start	uhrzeit	quittierung
2021-01-01	XYZ	08:00:00	2021-01-01 09:00:00
2021-01-01	XYZ	07:05:00	2021-01-01 09:05:00
2021-01-01	XYZ	07:05:00	2021-01-01 09:05:00
2021-01-01	XYZ	10:00:00	2021-01-01 10:05:00
2021-01-01	XYZ	10:00:00	2021-01-01 11:00:00

df = SPARK_DATAFRAME

def func1(x):
    count = df.filter(df.datum == x.datum).filter(df.start == x.start).filter(df.uhrzeit < x.uhrzeit).filter(df.quittierung[12:19] > x.uhrzeit).count()
    return (x.datum, x.start, count)



rdd2=df.rdd.map(lambda x: (func1(x))) 
df2=rdd2.toDF(["datum", "start", "count"])
df2.show()

通过这样做，我收到以下错误消息：

PicklingError: 无法序列化对象：TypeError: can't pickle _thread.RLock objects

有人可以帮助我如何将原始数据帧扩展为某些列，其中值基于对完整数据帧的查询？

示例结果

datum	start	uhrzeit	quittierung	count
2021-01-01	XYZ	08:00:00	2021-01-01 09:00:00	2
2021-01-01	XYZ	07:05:00	2021-01-01 09:05:00	0
2021-01-01	XYZ	07:06:00	2021-01-01 09:05:00	1
2021-01-01	XYZ	10:00:00	2021-01-01 10:04:00	0
2021-01-01	XYZ	10:05:00	2021-01-01 11:00:00	0

【问题讨论】：

标签： python pyspark

【解决方案1】：

看起来你只需要df.groupby("datum", "start").count().show() 不会给你想要的结果吗？还是真的需要使用rdd.map？

错误的原因是 spark 尝试序列化函数 func1 但是在 func1 中你有原始数据帧，它是不可序列化的。

【讨论】：

为简单起见，我删除了部分过滤连接。我认为我错过了关键细节。实际上，我不仅需要过滤 groupby 可以解决问题的相等值，而且还需要进行比较。你知道如何在没有 RDD 的情况下做到这一点吗？我尝试了 pandas 解决方法，但计算时间太长。
你能举一个结果数据框的例子吗？
我编辑了原始帖子

【解决方案2】：

您可以尝试使用 group by 和 Pandas UDF，如下所示：

df = SPARK_DATAFRAME

@pandas_udf(df.schema, functionType=PandasUDFType.GROUPED_MAP)
def func1(pdf):
    count = pdf.loc["same filtering but in Pandas"].count()
    return (count)


Result= df.groupBy(['datum', 'start']).apply(func1)

这个方向的东西应该可以工作

【讨论】：