pyspark 生成唯一值的所有组合

Posted 2023-02-25

【中文标题】pyspark 生成唯一值的所有组合

【英文标题】：pyspark generate all combinations of unique values

【发布时间】：2018-11-13 10:37:14

【问题描述】：

我正在尝试在我的 spark 数据框中生成唯一值的所有组合。 我想到的解决方案需要使用 itertools.product 和 pandas 数据框，因此效率不够。 这是我的代码：

all_date = [ i.Date for i in df.select("Date").distinct().collect()]
all_stores_id = [i.ID for i in fd.select("ID").distinct().collect()]
all_category = [i.CATEGORY for i in fd.select("CATEGORY").distinct().collect()]
combined = [all_date, all_stores_id, all_category]
all_combination_pdf= pd.DataFrame(columns = ['Date', 'ID', 'CATEGORY'], data=list(itertools.product(*combined)))
# convert pandas dataframe to spark
all_combination_df = sqlContext.createDataFrame(all_combination_pdf)
joined =  all_combination_df.join(df,["Date","ID","CATEGORY"],how="left")

有什么办法可以将此代码更改为更活泼的代码吗？

======编辑======

我也尝试使用 crossJoin 函数来实现这些功能。 下面是代码：

test_df = ((df.select('Date').distinct()).crossJoin(df.select('ID').distinct())).crossJoin(df.select('CATEGORY').distinct())
test_df.show(10)

由于某种未知原因引发以下异常：

An error occurred while calling o305.showString.
: java.lang.OutOfMemoryError: GC overhead limit exceeded
    at java.lang.Integer.valueOf(Integer.java:832)

【问题讨论】：

“一些未知原因” - 这个错误很明显。你的内存不足了。你有多少不同的价值观？

unknown 这个词的选择很差。我知道错误是由于内存限制造成的，但我不知道为什么会发生。数据样本生成大约 1M 不同的值，更重要的是，使用 pandas 实现的代码运行良好。您知道如何将 pandas 代码重新实现为高效的 p​​yspark 代码吗？

【参考方案1】：

您可以使用它生成数据框。它只是使用每列的唯一值创建一个数据框，并与其他列执行交叉连接（笛卡尔积）。

((df.select('Date').distinct()).crossJoin(df.select('ID').distinct())).crossJoin(df.select('CATEGORY').distinct())

它可以放在一个 for 循环中，通过一些工作来为其他数据帧自动化它。

希望对你有帮助

【讨论】：

感谢您的回答。不幸的是，出于某种我不知道的原因，我无法执行 test_df.show(10) 之后的行。我得到 java.lang.OutOfMemoryError: GC 开销限制超出 的错误。项目是在Azure平台上开发的，所以这绝对不是硬件问题。

抱歉没有帮助。 Azure 中的内存错误肯定很奇怪，但想想将要创建的巨大数据帧。例如，如果您有 3 列，每列有 5 个不同的值，那么您最终将得到 5^3 = 125 行。想象更大的价值。