基于列条件的PySpark dropdups

Question

仍然是新的火花，我正试图做这最后的转变尽可能干净和有效。你知道吗

假设我有一个如下所示的数据帧

+------+--------+                  
|ID    | Hit    |                  
+------+--------+
|123   |   0    | 
|456   |   1    |
|789   |   0    |     
|123   |   1    |   
|123   |   0    | 
|789   |   1    |   
|1234  |   0    |
| 1234 |   0    |   
+------+--------+

我试着以一个新的数据帧结束（或者两个，取决于什么更有效），如果一行在“hit”中有一个1，它就不能有一个在hit中有一个0的行，如果有，0将是基于ID列的不同级别。你知道吗

这是我尝试过的方法之一，但我不确定这是否正确 1最有效的方法 2最干净的方法

dfhits = df.filter(df.Hit == 1)
dfnonhits = df.filter(df.Hit == 0)
dfnonhitsdistinct = dfnonhits.filter(~dfnonhits['ID'].isin(dfhits))

Enddataset将如下所示：

+------+--------+                  
|ID    | Hit    |                  
+------+--------+
|456   |   1    |    
|123   |   1    |   
|789   |   1    |   
|1234  |   0    |  
+------+--------+

Answer 1

# Creating the Dataframe.
from pyspark.sql.functions import col
df = sqlContext.createDataFrame([(123,0),(456,1),(789,0),(123,1),(123,0),(789,1),(500,0),(500,0)],
                                ['ID','Hit']) 
df.show()
+ -+ -+ 
| ID|Hit| 
+ -+ -+ 
|123|  0| 
|456|  1| 
|789|  0| 
|123|  1| 
|123|  0| 
|789|  1| 
|500|  0| 
|500|  0| 
+ -+ -+

这个想法是找到每个ID中total的Hit，如果它大于0，则意味着Hit中至少存在一个1。因此，当这个条件为true时，我们将删除所有rows值为Hit的rows。你知道吗

# Registering the dataframe as a temporary view.
df.registerTempTable('table_view')
df=sqlContext.sql(
    'select ID, Hit, sum(Hit) over (partition by ID) as sum_Hit from table_view'
)
df.show()
+ -+ -+   -+ 
| ID|Hit|sum_Hit| 
+ -+ -+   -+ 
|789|  0|      1| 
|789|  1|      1| 
|500|  0|      0| 
|500|  0|      0| 
|123|  0|      1| 
|123|  1|      1| 
|123|  0|      1| 
|456|  1|      1| 
+ -+ -+   -+
df = df.filter(~((col('Hit')==0) & (col('sum_Hit')>0))).drop('sum_Hit').dropDuplicates()
df.show()
+ -+ -+ 
| ID|Hit|  
+ -+ -+ 
|789|  1| 
|500|  0| 
|123|  1| 
|456|  1|
+ -+ -+