PySpark drop-dupes основан на условии столбца

Question

Все еще новичок в Spark, и я пытаюсь сделать это окончательное преобразование максимально чистым и эффективным.

Скажем, у меня есть кадр данных, который выглядит следующим образом

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

Iя пытаюсь получить новый фрейм данных (или два, в зависимости от того, что более эффективно), где, если строка имеет «1» в «попадании», она не может иметь строку с «0» в попадании, а если есть, то «0»было бы до определенного уровня, основанного на столбце ID.

Вот один из методов, которые я пробовал, но я не уверен, что это 1. Самый эффективный из возможных способов 2. Самый чистый возможный способ

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

Конечный набор данных будет выглядеть следующим образом:

+------+--------+                  
|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| 
+---+---+

Идея состоит в том, чтобы найти total из Hit на ID, и если оно больше 0, это означает, что в Hit присутствует хотя бы один 1.Таким образом, когда это условие true, мы удалим все rows с Hit значениями 0.

# 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|
+---+---+