python - tutorial - Obtenga los distintos elementos de cada grupo por otro campo en un Spark 1.6 Dataframe

sqlcontext spark python

Question

Estoy intentando agrupar por fecha en un marco de datos de Spark y para cada grupo contar los valores únicos de una columna:

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}

Y el código:

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()

Los resultados con pyspark son

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

Y lo que estoy esperando es algo como esto con los pandas:

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

¿Cómo puedo obtener los elementos únicos de cada grupo por otro campo, como la dirección?

Answer 1

Hay una manera de hacer este recuento de elementos distintos de cada grupo usando la función 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|
+----+--------------+

Los documentos están disponibles [aquí] ( https://spark.apache.org/docs/1.6.0/api/java/org/apache/spark/sql/functions.html#countDistinct(org.apache.spark.sql. Column , org.apache.spark.sql.Column ...)).