Récupère les éléments distincts de chaque groupe par un autre champ sur un Dataframe Spark 1.6

Je suis en train de groupe par date dans une Étincelle dataframe et pour chaque groupe de compter les valeurs uniques d'une colonne:

test.json
{"name":"Yin", "address":1111111, "date":20151122045510}
{"name":"Yin", "address":1111111, "date":20151122045501}
{"name":"Yln", "address":1111111, "date":20151122045500}
{"name":"Yun", "address":1111112, "date":20151122065832}
{"name":"Yan", "address":1111113, "date":20160101003221}
{"name":"Yin", "address":1111111, "date":20160703045231}
{"name":"Yin", "address":1111114, "date":20150419134543}
{"name":"Yen", "address":1111115, "date":20151123174302}

Et le code:

import pyspark.sql.funcions as func
from pyspark.sql.types import TimestampType
from datetime import datetime

df_y = sqlContext.read.json("/user/test.json")
udf_dt = func.udf(lambda x: datetime.strptime(x, '%Y%m%d%H%M%S'), TimestampType())
df = df_y.withColumn('datetime', udf_dt(df_y.date))
df_g = df_y.groupby(func.hour(df_y.date))    
df_g.count().distinct().show()

Les résultats avec pyspark sont

df_y.groupby(df_y.name).count().distinct().show()
+----+-----+
|name|count|
+----+-----+
| Yan|    1|
| Yun|    1|
| Yin|    4|
| Yen|    1|
| Yln|    1|
+----+-----+

Et que j'attends est quelque chose comme ça avec les pandas:

df = df_y.toPandas()
df.groupby('name').address.nunique()
Out[51]: 
name
Yan    1
Yen    1
Yin    2
Yln    1
Yun    1

Comment puis-je obtenir les éléments uniques de chaque groupe par un autre domaine, telles que l'adresse?

source d'informationauteur Ivan

  1. 21

    Il y a un moyen de faire ce nombre d'éléments distincts de chaque groupe à l'aide de la fonction countDistinct:

    import pyspark.sql.functions as func
from pyspark.sql.types import TimestampType
from datetime import datetime

df_y = sqlContext.read.json("/user/test.json")
udf_dt = func.udf(lambda x: datetime.strptime(x, '%Y%m%d%H%M%S'), TimestampType())
df = df_y.withColumn('datetime', udf_dt(df_y.date))
df_g = df_y.groupby(func.hour(df_y.date))    
df_y.groupby(df_y.name).agg(func.countDistinct('address')).show()

+----+--------------+
|name|count(address)|
+----+--------------+
| Yan|             1|
| Yun|             1|
| Yin|             2|
| Yen|             1|
| Yln|             1|
+----+--------------+

    Les docs sont disponibles [ici](https://spark.apache.org/docs/1.6.0/api/java/org/apache/spark/sql/functions.html#countDistinct(org.apache.spark.sql.Columnorg.apache.spark.sql.Colonne...)).