title: How to use countDistinct in Scala with Spark?
body: <p>I've tried to use <em>countDistinct</em> function which should be available in Spark 1.5 according to <a href="https://databricks.com/blog/2015/09/16/spark-1-5-dataframe-api-highlights-datetimestring-handling-time-intervals-and-udafs.html" rel="noreferrer">DataBrick's blog</a>. However, I got the following exception:</p>
  
  <pre><code>Exception in thread "main" org.apache.spark.sql.AnalysisException: undefined function countDistinct;
  </code></pre>
  
  <p>I've found that on <a href="http://apache-spark-developers-list.1001551.n3.nabble.com/Exception-when-using-some-aggregate-operators-td14795.html" rel="noreferrer">Spark developers' mail list</a> they suggest using <em>count</em> and <em>distinct</em> functions to get the same result which should be produced by <em>countDistinct</em>:</p>
  
  <pre><code>count(distinct &lt;columnName&gt;)
  // Instead
  countDistinct(&lt;columnName&gt;)
  </code></pre>
  
  <p>Because I build aggregation expressions dynamically from the list of the names of aggregation functions I'd prefer to don't have any special cases which require different treating.</p>
  
  <p>So, is it possible to unify it by:</p>
  
  <ul>
  <li>registering new UDAF which will be an alias for <em>count(distinct columnName)</em></li>
  <li><p>registering manually already implemented in Spark CountDistinct function which is probably one from following import:</p>
  
  <p>import org.apache.spark.sql.catalyst.expressions.{CountDistinctFunction, CountDistinct}</p></li>
  <li><p>or do it in any other way?</p></li>
  </ul>
  
  <p>EDIT:
  Example (with removed some local references and unnecessary code):</p>
  
  <pre><code>import org.apache.spark.SparkContext
  import org.apache.spark.sql.{Column, SQLContext, DataFrame}
  import org.apache.spark.sql.functions._
  
  import scala.collection.mutable.ListBuffer
  
  
  class Flattener(sc: SparkContext) {
    val sqlContext = new SQLContext(sc)
  
    def flatTable(data: DataFrame, groupField: String): DataFrame = {
      val flatteningExpressions = data.columns.zip(TypeRecognizer.getTypes(data)).
        flatMap(x =&gt; getFlatteningExpressions(x._1, x._2)).toList
  
      data.groupBy(groupField).agg (
        expr(s"count($groupField) as groupSize"),
        flatteningExpressions:_*
      )
    }
  
    private def getFlatteningExpressions(fieldName: String, fieldType: DType): List[Column] = {
      val aggFuncs = getAggregationFunctons(fieldType)
  
      aggFuncs.map(f =&gt; expr(s"$f($fieldName) as ${fieldName}_$f"))
    }
  
    private def getAggregationFunctons(fieldType: DType): List[String] = {
      val aggFuncs = new ListBuffer[String]()
  
      if(fieldType == DType.NUMERIC) {
        aggFuncs += ("avg", "min", "max")
      }
  
      if(fieldType == DType.CATEGORY) {
        aggFuncs += "countDistinct"
      }
  
      aggFuncs.toList
    }
  
  }
  </code></pre>