Pyspark 数据框转换为 pandas 会丢弃数据？答案

【问题标题】：Pyspark dataframe conversion to pandas drops data?Pyspark 数据框转换为 pandas 会丢弃数据？
【发布时间】：2018-05-08 04:09:03
【问题描述】：

我有一个相当复杂的过程来创建 pyspark 数据框，将其转换为 pandas 数据框，并将结果输出到平面文件。不知道是在什么时候引入了错误，所以我将描述整个过程。

一开始我有一个 pyspark 数据框，其中包含 id 集的成对相似性。它看起来像这样：

  +------+-------+-------------------+
  |  ID_A|   ID_B|  EuclideanDistance|
  +------+-------+-------------------+
  |     1|      1|                0.0|
  |     1|      2|0.13103884200454394|
  |     1|      3| 0.2176246463836219|
  |     1|      4|  0.280568636550471|
 ...

我喜欢按 ID_A 对其进行分组，按 EuclideanDistance 对每个组进行排序，并且只获取每个组的前 N 对。所以首先我这样做：

from pyspark.sql.window import Window
from pyspark.sql.functions import rank, col, row_number

window = Window.partitionBy(df['ID_A']).orderBy(df_sim['EuclideanDistance'])
result = (df.withColumn('row_num', row_number().over(window)))

我确保 ID_A = 1 仍在“结果”数据框中。然后我这样做是为了将每个组限制为 20 行：

result1 = result.where(result.row_num<20)
result1.toPandas().to_csv("mytest.csv")

并且 ID_A = 1 不在生成的 .csv 文件中（尽管它仍然存在于 result1 中）。在这个转换链中的某个地方是否存在可能导致数据丢失的问题？

【问题讨论】：

标签： pandas pyspark

【解决方案1】：

您在解决方案的窗口中引用了 2 个数据框。不确定这是否会导致您的错误，但值得清理。在任何情况下，您都不需要在 window definition 中引用特定的数据框。无论如何，请尝试

window = Window.partitionBy('ID_A').orderBy('EuclideanDistance')

【讨论】：

【解决方案2】：

正如大卫所说，您在窗口函数中引用了第二个数据框“df_sim”。

我测试了以下内容，它可以在我的机器上运行（著名的遗言）：

from pyspark.sql.window import Window
from pyspark.sql.functions import rank, col, row_number
import pandas as pd

#simulate some data
df = pd.DataFrame({'ID_A': pd.np.arange(100)%5, 
    'ID_B': pd.np.repeat(pd.np.arange(20),5), 
    'EuclideanDistance': pd.np.random.rand(100)*5}
    )
#artificially set distance between point and self to 0
df['EuclideanDistance'][df['ID_A'] == df['ID_B']] = 0
df = spark.createDataFrame(df)
#end simulation
window = Window.partitionBy(df['ID_A']).orderBy(df['EuclideanDistance'])
output = df.select('*', row_number().over(window).alias('rank')).filter(col('rank') <= 10)
output.show(50)

模拟代码只是为了使它成为一个独立的示例。您当然可以使用您的实际数据框并在测试时忽略模拟。希望有效！

【讨论】：