Pyspark：基於

Question

我'使用pyspark每行的空值的數量過濾數據幀我有這樣的一個表：

id | ClientNum | Value |  Date  | Age | Country | Job 
1 |  19  | A | 1483695000 | 21 | null | null 
2 |  19  | A | 1483696500 | 21 | France | null 
3 |  19  | A | 1483697800 | 21 | France | Engineer 
4 |  19  | B | 1483699000 | 21 | null | null 
5 |  19  | B | 1483699500 | 21 | France | null 
6 |  19  | B | 1483699800 | 21 | France | Engineer 
7 |  24  | C | 1483699200 | null | null | null 
8 |  24  | D | 1483699560 | 28 | Spain | null 
9 |  24  | D | 1483699840 | 28 | Spain | Student 

基於列價值，我想保持每個ClientNum不同值指定了大多數信息（年齡，國家，工作）。

的結果應該是這樣的：

ClientNum | Value |  Date  | Age | Country | Job 
     19  | A | 1483697800 | 21 | France | Engineer 
     19  | B | 1483699800 | 21 | France | Engineer 
     24  | C | 1483699200 | null | null | null 
     24  | D | 1483699840 | 28 | Spain | Student 

謝謝！

Answer 1

下面是使用udf計算每行非空值的數量的方法，以及隨後使用Window功能篩選數據：

讓我們先來定義udf以列的array作爲參數，並給出了結果的非空值的數量。

from pyspark.sql.functions import array 

def nullcounter(arr): 

    res = [x for x in arr if x != None] 
    return(len(res)) 

nullcounter_udf = udf(nullcounter) 

讓這列添加到您的數據：

df = df.withColumn("counter", nullcounter_udf(array(df.columns))) 

現在我們可以通過ClientNum和Value劃分您的數據，並保持行最高counter值：

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

window = Window.partitionBy(df['ClientNum'], df['Value']).orderBy(df['counter'].desc()) 

df.select('*', rank().over(window).alias('rank')) \ 
    .filter(col('rank') == 1) \ 
    .sort('Value') \ 
    .show() 
+---+---------+-----+----------+----+-------+--------+-------+----+ 
| id|ClientNum|Value|  Date| Age|Country|  Job|counter|rank| 
+---+---------+-----+----------+----+-------+--------+-------+----+ 
| 3|  19| A|1483697800| 21| France|Engineer|  8| 1| 
| 6|  19| B|1483699800| 21| France|Engineer|  8| 1| 
| 7|  24| C|1483699200|null| null| null|  5| 1| 
| 9|  24| D|1483699840| 28| Spain| Student|  8| 1| 
+---+---------+-----+----------+----+-------+--------+-------+----+ 

數據

df = sc.parallelize([(1, 19, "A", 1483695000, 21, None, None), 
(2, 19, "A", 1483696500, 21, "France", None), 
(3, 19, "A", 1483697800, 21, "France", "Engineer"), 
(4, 19, "B", 1483699000, 21, None, None), 
(5, 19, "B", 1483699500, 21, "France", None), 
(6, 19, "B", 1483699800, 21, "France", "Engineer"), 
(7, 24, "C", 1483699200, None, None, None), 
(8, 24, "D", 1483699560, 28, "Spain", None), 
(9, 24, "D", 1483699840, 28, "Spain", "Student")]).toDF(["id","ClientNum","Value","Date","Age", "Country", "Job"]) 

Answer 2

試試這個：

val df = Your_data_frame.registerTempTable("allData") // register your dataframe as a temp table 

// we are finding max of date for each clientNum and value and join back to the original table. 

    sqlContext.sql("select a.ClientNum, a.Value, a.Date, a.Age, a.Country, a.Job from allData a 
    join 
    (select ClientNum, Value, max(Date) as max_date from allData group by ClientNum, Value) b 
    on a.ClientNum = b.ClientNum and a.Value = b.Value and a.Date = b.max_date").show 

Answer 3

如果像我一樣，你有沒有和其他答案的煩惱，下面是一個使用UDF我在Python溶液（火花2.2.0）：

讓我們創建一個虛擬數據集：

llist = [(1, 'alice', 'some_field', 'some_field', 'some_field', None), (30, 'bob', 'some_field', None, None, 10), (3, 'charles', 'some_field', None, 'some_other_field', 1111)] 
df = sqlContext.createDataFrame(llist, ['id', 'name','field1','field2', 'field3', 'field4']) 

df.show() 

+---+-------+----------+----------+----------------+------+ 
| id| name| field1| field2|   field3|field4| 
+---+-------+----------+----------+----------------+------+ 
| 1| alice|some_field|some_field|  some_field| null| 
| 30| bob|some_field|  null|   null| 10| 
| 3|charles|some_field|  null|some_other_field| 1111| 
+---+-------+----------+----------+----------------+------+ 

讓我們來定義UDF計數None值：

from pyspark.sql.types import IntegerType 
from pyspark.sql.functions import struct, udf 

count_empty_columns = udf(
         lambda row: len([x for x in row if x is None]), 
         IntegerType() 
        ) 

我們可以基於UDF添加新列null_count：

df = df.withColumn('null_count', 
     count_empty_columns(struct([df[x] for x in df.columns]))) 

df.show() 

+---+-------+----------+----------+----------------+------+----------+ 
| id| name| field1| field2|   field3|field4|null_count| 
+---+-------+----------+----------+----------------+------+----------+ 
| 1| alice|some_field|some_field|  some_field| null|   1| 
| 30| bob|some_field|  null|   null| 10|   2| 
| 3|charles|some_field|  null|some_other_field| 1111|   1| 
+---+-------+----------+----------+----------------+------+----------+ 

最後篩選：

df = df.filter(df['null_count'] <= 1)