title: filter KeyValueGrouped Dataset in spark
body: <p>I have a typed dataset of a custom class and use groupbykey method on it. You know that it results a KeyValueGroupedDataset. I want to filter this new dataset but there is no filter method for this type of dataset. So, My question is: How can I filter on this type of dataset? (Java solution is needed. spark version: 2.3.1). </p>
  
  <p>sampleData:</p>
  
  <pre><code>"id":1,"fname":"Gale","lname":"Willmett","email":"gwillmett0@nhs.uk","gender":"Female"
  "id":2,"fname":"Chantalle","lname":"Wilcher","email":"cwilcher1@blinklist.com","gender":"Female"
  "id":3,"fname":"Polly","lname":"Grandisson","email":"pgrandisson2@linkedin.com","gender":"Female"
  "id":3,"fname":"Moshe","lname":"Pink","email":"mpink3@twitter.com","gender":"Male"
  "id":2,"fname":"Yorke","lname":"Ginnelly","email":"yginnelly4@apple.com","gender":"Male"
  </code></pre>
  
  <p>And What I did:</p>
  
  <pre><code>    Dataset&lt;Person&gt; peopleDS = spark.read().format("parquet").load("\path").as(Encoders.bean(Person.class));
      KeyValueGroupedDataset&lt;String, Person&gt; KVDS = peopleDS.groupByKey( (MapFunction&lt;Person, String&gt; ) f -&gt; f.getGender() , Encoders.STRING());
  //How Can I filter on KVDS's id field? 
  </code></pre>
  
  <p><strong>Update1</strong> (use of flatMapGroups):</p>
  
  <pre><code>Dataset&lt;Person&gt; persons = KVDS.flatMapGroups((FlatMapGroupsFunction &lt;String,Person,Person&gt;) (f,k) -&gt; (Iterator&lt;Person&gt;) k ,  Encoders.bean(Person.class));
  </code></pre>
  
  <p><strong>Update2</strong> (use of MapGroups)</p>
  
  <pre><code>Dataset&lt;Person&gt; peopleMap = KVDS.mapGroups((MapGroupsFunction &lt;String,Person,Person&gt;) (f,g) -&gt; {
          while (g.hasNext()) {
          //What can I do here?       
      }
  },Encoders.bean(Person.Class); 
  </code></pre>
  
  <p><strong>Update3</strong> : I want to filter those groups that distinct of their ids is greater than 1. for example in below picture: I want just Female groups because distinct of their ids is greater that 1 (first field is id. Others are fname,lname,email and gender).
  <a href="https://i.stack.imgur.com/jSpYs.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/jSpYs.png" alt="enter image description here"></a></p>
  
  <p><strong>Update4:</strong> I did What I want with "RDD", but I want to do exactly this part of code with "Dataset":</p>
  
  <pre><code>List&lt;Tuple2&lt;String, Iterable&lt;Person&gt;&gt;&gt; f = PersonRDD
          .mapToPair(s -&gt; new Tuple2&lt;&gt;(s.getGender(), s)).groupByKey()
          .filter(t -&gt; ((Collection&lt;Person&gt;) t._2()).stream().mapToInt(e -&gt; e.getId).distinct().count() &gt; 1)
          .collect();
  </code></pre>