title: Spark: How to translate count(distinct(value)) in Dataframe API's
body: <p>I'm trying to compare different ways to aggregate my data.</p>
  <p>This is my input data with 2 elements (page,visitor):</p>
  <pre><code>(PAG1,V1)
  (PAG1,V1)
  (PAG2,V1)
  (PAG2,V2)
  (PAG2,V1)
  (PAG1,V1)
  (PAG1,V2)
  (PAG1,V1)
  (PAG1,V2)
  (PAG1,V1)
  (PAG2,V2)
  (PAG1,V3)
  </code></pre>
  <p>Working with a SQL command into Spark SQL with this code:</p>
  <pre><code>import sqlContext.implicits._
  case class Log(page: String, visitor: String)
  val logs = data.map(p =&gt; Log(p._1,p._2)).toDF()
  logs.registerTempTable(&quot;logs&quot;)
  val sqlResult= sqlContext.sql(
                                &quot;&quot;&quot;select page
                                         ,count(distinct visitor) as visitor
                                     from logs
                                 group by page
                                &quot;&quot;&quot;)
  val result = sqlResult.map(x=&gt;(x(0).toString,x(1).toString))
  result.foreach(println)
  </code></pre>
  <p>I get this output:</p>
  <pre><code>(PAG1,3) // PAG1 has been visited by 3 different visitors
  (PAG2,2) // PAG2 has been visited by 2 different visitors
  </code></pre>
  <p>Now, I would like to get the same result using Dataframes and thiers API, but I can't get the same output:</p>
  <pre><code>import sqlContext.implicits._
  case class Log(page: String, visitor: String)
  val logs = data.map(p =&gt; Coppia(p._1,p._2)).toDF()
  val result = log.select(&quot;page&quot;,&quot;visitor&quot;).groupBy(&quot;page&quot;).count().distinct
  result.foreach(println)
  </code></pre>
  <p>In fact, that's what I get as output:</p>
  <pre><code>[PAG1,8]  // just the simple page count for every page
  [PAG2,4]
  </code></pre>