Spark: Dataframe API'sının sayımı (ayrı (değer))

Question

Verilerimi birleştirmenin farklı yollarını karşılaştırmaya çalışıyorum. Bu kod ile Spark SQL içine bir SQL komutu ile çalışma

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

:

Bu

2 elementlerin (sayfa, ziyaretçi) ile benim girdi veridir

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) 

bu çıktıyı almak:

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

Şimdi, Dataframes ve thiers API kullanarak aynı sonucu elde etmek istiyorum, ancak aynı şeyi alamıyorum. çıkışı:

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

Muhtemelen dilsiz bir şey, ama şu anda göremiyorum: Aslında

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) 

, ben çıktı olarak elde ediyoruz.

Şimdiden teşekkürler!

FF

Answer 1

Ne gerek DataFrame toplama fonksiyonu countDistinct: Ben bu hatayı alıyorum

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)