Spark: Come convertire count (distinto (valore)) in Dataframe API

Question

Sto cercando di confrontare diversi modi per aggregare i miei dati.

Questo è il mio dati di input con 2 elementi (pagina, visitatori):

(PAG1,V1) 
(PAG1,V1) 
(PAG2,V1) 
(PAG2,V2) 
(PAG2,V1) 
(PAG1,V1) 
(PAG1,V2) 
(PAG1,V1) 
(PAG1,V2) 
(PAG1,V1) 
(PAG2,V2) 
(PAG1,V3) 

Lavorare con un comando SQL in Spark SQL con questo codice:

import sqlContext.implicits._ 
case class Log(page: String, visitor: String) 
val logs = data.map(p => Log(p._1,p._2)).toDF() 
logs.registerTempTable("logs") 
val sqlResult= sqlContext.sql(
           """select page 
             ,count(distinct visitor) as visitor 
            from logs 
           group by page 
           """) 
val result = sqlResult.map(x=>(x(0).toString,x(1).toString)) 
result.foreach(println) 

ottengo questo output:

(PAG1,3) // PAG1 has been visited by 3 different visitors 
(PAG2,2) // PAG2 has been visited by 2 different visitors 

Ora, mi piacerebbe ottenere lo stesso risultato utilizzando Dataframes e l'API di thiers, ma non riesco a ottenere lo stesso Uscita:

import sqlContext.implicits._ 
case class Log(page: String, visitor: String) 
val logs = data.map(p => Coppia(p._1,p._2)).toDF() 
val result = log.select("page","visitor").groupBy("page").count().distinct 
result.foreach(println) 

In realtà, questo è quello che ottengo come output:

[PAG1,8] // just the simple page count for every page 
[PAG2,4] 

E 'probabilmente qualcosa di stupido, ma non riesco a vedere in questo momento.

Grazie in anticipo!

FF

Answer 1

Quello che vi serve è la funzione di aggregazione dataframe countDistinct:

import sqlContext.implicits._ 
import org.apache.spark.sql.functions._ 

case class Log(page: String, visitor: String) 

val logs = data.map(p => Coppia(p._1,p._2)) 
      .toDF() 

val result = log.select("page","visitor") 
      .groupBy('page) 
      .agg('page, countDistinct('visitor)) 

result.foreach(println)