title: Counting distinct values for a given column partitioned by a window function, without using approx_count_distinct()
body: <p>I have the following dataframe:</p>
  <pre><code>val df1 = Seq((&quot;Roger&quot;,&quot;Rabbit&quot;, &quot;ABC123&quot;), (&quot;Roger&quot;,&quot;Rabit&quot;, &quot;ABC123&quot;),(&quot;Roger&quot;,&quot;Rabbit&quot;, &quot;ABC123&quot;), (&quot;Trevor&quot;,&quot;Philips&quot;,&quot;XYZ987&quot;), (&quot;Trevor&quot;,&quot;Philips&quot;,&quot;XYZ987&quot;)).toDF(&quot;first_name&quot;, &quot;last_name&quot;, &quot;record&quot;)
  
  +----------+---------+------+
  |first_name|last_name|record|
  +----------+---------+------+
  |Roger     |Rabbit   |ABC123|
  |Roger     |Rabit    |ABC123|
  |Roger     |Rabbit   |ABC123|
  |Trevor    |Philips  |XYZ987|
  |Trevor    |Philips  |XYZ987|
  +----------+---------+------+
  </code></pre>
  <p>I want to group records in this dataframe by the column <code>record</code>. And then I want to look for anomalies in the fields <code>first_name</code> and <code>last_name</code>, which should remain constant for all records with same <code>record</code> value.</p>
  <p>The best approach I found so far is using <code>approx_count_distinct</code>:</p>
  <pre><code>val wind_person = Window.partitionBy(&quot;record&quot;)
  
  df1.withColumn(&quot;unique_fields&quot;,cconcat($&quot;first_name&quot;,$&quot;last_name&quot;))
  .withColumn(&quot;anomaly&quot;,capprox_count_distinct($&quot;unique_fields&quot;) over wind_person)
  .show(false)    
  
  
  +----------+---------+------+-------------+-------+
  |first_name|last_name|record|unique_fields|anomaly|
  +----------+---------+------+-------------+-------+
  |Roger     |Rabbit   |ABC123|RogerRabbit  |2      |
  |Roger     |Rabbit   |ABC123|RogerRabbit  |2      |
  |Roger     |Rabit    |ABC123|RogerRabit   |2      |
  |Trevor    |Philips  |XYZ987|TrevorPhilips|1      |
  |Trevor    |Philips  |XYZ987|TrevorPhilips|1      |
  +----------+---------+------+-------------+-------+
  </code></pre>
  <p>Where an anomaly is detected is <code>anomaly</code> column is greater than 1.</p>
  <p>The problem is with <code>approx_count_distinct</code> we get just an approximation, and I am not sure how much confident we can be that it will always return an accurate count.</p>
  <p>Some extra information:</p>
  <ul>
  <li>The Dataframe may contain over 500M records</li>
  <li>The Dataframe is previously repartitioned based on <code>record</code> column</li>
  <li>For each different value of <code>record</code>, no more than 15 rows will be there</li>
  </ul>
  <p>Is is safe to use <code>approx_count_distinct</code> in this scenario with a 100% accuracy or are there better window functions in spark to achieve this?</p>