Récupère les premières valeurs non nulles dans le groupe (Spark 1.6)

Question

Comment puis-je obtenir les premières valeurs non nulles d'un groupe? J'ai essayé d'utiliser first avec coalesceF.first(F.coalesce("code")) mais je n'obtiens pas le comportement désiré (je semble obtenir la première rangée).

from pyspark import SparkContext 
from pyspark.sql import SQLContext 
from pyspark.sql import functions as F 

sc = SparkContext("local") 

sqlContext = SQLContext(sc) 

df = sqlContext.createDataFrame([ 
    ("a", None, None), 
    ("a", "code1", None), 
    ("a", "code2", "name2"), 
], ["id", "code", "name"]) 

J'ai essayé:

(df 
    .groupby("id") 
    .agg(F.first(F.coalesce("code")), 
     F.first(F.coalesce("name"))) 
    .collect()) 

SORTIE DÉSIRÉ

[Row(id='a', code='code1', name='name2')] 

Answer 1

Pour Spark 01.03 à 01.05, cela pourrait faire l'affaire:

from pyspark.sql import functions as F 
df.groupBy(df['id']).agg(F.first(df['code']), F.first(df['name'])).show() 

+---+-----------+-----------+ 
| id|FIRST(code)|FIRST(name)| 
+---+-----------+-----------+ 
| a|  code1|  name2| 
+---+-----------+-----------+ 

Édition

Apparemment, dans la version 1.6, ils ont changé la façon dont la fonction d'agrégat first est traitée. Maintenant, la classe sous-jacente First doit être construite avec un second paramètre ignoreNullsExpr, qui n'est pas encore utilisé par la fonction d'agrégation first (comme cela peut être vu here). Cependant, dans Spark 2.0, il pourra appeler le agg(F.first(col, True)) pour ignorer les valeurs nulles (comme cela peut être vérifié here). Par conséquent, pour Spark 1.6, l'approche doit être différente et un peu plus inefficace, de façon imprévisible. Une idée est la suivante:

from pyspark.sql import functions as F 
df1 = df.select('id', 'code').filter(df['code'].isNotNull()).groupBy(df['id']).agg(F.first(df['code'])) 
df2 = df.select('id', 'name').filter(df['name'].isNotNull()).groupBy(df['id']).agg(F.first(df['name'])) 
result = df1.join(df2, 'id') 
result.show() 

+---+-------------+-------------+ 
| id|first(code)()|first(name)()| 
+---+-------------+-------------+ 
| a|  code1|  name2| 
+---+-------------+-------------+ 

Peut-être qu'il existe une meilleure option. Je vais modifier la réponse si j'en trouve un.

Answer 2

Parce que je n'avait qu'une seule valeur non nulle pour chaque groupe, en utilisant min/max à 1,6 a travaillé pour mes besoins:

(df 
    .groupby("id") 
    .agg(F.min("code"), 
     F.min("name")) 
    .show()) 

+---+---------+---------+ 
| id|min(code)|min(name)| 
+---+---------+---------+ 
| a| code1| name2| 
+---+---------+---------+