title: Kafka stream count distinct in scala?
body: <p>I need to do a distinct count per user in a kafka stream. This is my initial implementation but has the error on aggregate</p>
  
  <pre><code>Required: [Seq[String], mutableHashSet[String]]
  Found: mutable.HashSet[String]
  </code></pre>
  
  <p>and I'm not sure how to provide a custom serde for <code>mutable.HashSet</code> either ...</p>
  
  <pre><code>val totalUniqueCategoriesCounts: KTable[String, Int] = inputStream
      .filter((_ , ev) =&gt; ev.eventData.evData.pageType.isDefined)
      .groupBy((_, ev) =&gt; ev.eventData.custData.customerUid.get)
      .aggregate(initializer = mutable.HashSet[String])(
        (aggKey: String, newValue: Event, aggValue: mutable.HashSet[String]) =&gt; {
          val cat: String = newValue.eventData.cntData.contentCategory.get
          aggValue += cat
          aggValue
        }, **Serde Here?**)
      .mapValues((set: mutable.HashSet[String]) =&gt; set.size)
      //.count()
    totalUniqueCategoriesCounts.toStream.to("total_unique_categories")
  </code></pre>
  
  <p>Any help would be appreciated.</p>
  
  <p>I'm also concerned about performance. Is this the best way to do a count distinct in a kafka stream?</p>
  
  <p><strong>Update</strong> Fixed the code issue but still concerned about the performance implications of this (if any) or any better ways to do the same thing.</p>