title: How can I use a custom CombineFn with Combine.GroupedValues?
body: <p>I've written a CombineFn that has input <code>KV&lt;String, TableRow&gt;</code> and output <code>KV&lt;String, Iterable&lt;TableRow&gt;&gt;</code>. I'd like to use Combine.GroupedValues (or Combine.PerKey), and the source could would seem to suggest that this is possible, but I'm getting the following error:</p>
  
  <pre><code>Incorrect number of type arguments for generic method &lt;K, V&gt;groupedValues(SerializableFunction&lt;Iterable&lt;V&gt;,V&gt;) of type Combine; it cannot be parameterized with arguments &lt;String, TableRow, Iterable&lt;TableRow&gt;&gt;
  </code></pre>
  
  <p>We're using Beam v2.10. The context here is that we're applying a session window to a PCollection of <code>KV&lt;String, TableRow&gt;</code>, then using a <code>GroupByKey</code> to create a PCollection of <code>KV&lt;String, Iterable&lt;TableRow&gt;&gt;</code>. After this step, our CombineFn reduces each group to <code>KV&lt;String, Iterable&lt;TableRow&gt;&gt;</code>, the Iterable containing TableRows created based on the contents of the input.</p>
  
  <p>The transform steps:</p>
  
  <pre class="lang-java prettyprint-override"><code>public PCollection&lt;KV&lt;String, Iterable&lt;TableRow&gt;&gt;&gt; expand(PCollection&lt;KV&lt;String, TableRow&gt;&gt; rows) {
  
    // group by step
    PCollection&lt;KV&lt;String, Iterable&lt;TableRow&gt;&gt;&gt; groupedValues = rows.apply(
      "Group by Key",
      GroupByKey.&lt;String, TableRow&gt;Create()
    );
  
      // combine step
    PCollection&lt;KV&lt;String, Iterable&lt;TableRow&gt;&gt;&gt; combinedValues = groupedValues.apply(
      "Generate New Rows",
      // errors here
      // Incorrect number of type arguments for generic 
      // method &lt;K, V&gt;groupedValues(SerializableFunction&lt;Iterable&lt;V&gt;,V&gt;) 
      // of type Combine; it cannot be parameterized with arguments 
      // &lt;String, TableRow, Iterable&lt;TableRow&gt;&gt;
      Combine.&lt;String, TableRow, Iterable&lt;TableRow&gt;&gt;groupedValues(new CreateEvents())
    );
  
  
    return combinedValues;
  }
  </code></pre>
  
  <p>Combine Function:</p>
  
  <pre class="lang-java prettyprint-override"><code>private static class CreateEvents extends CombineFn&lt;KV&lt;String, TableRow&gt;, CreateEvents.Accum, KV&lt;String, Iterable&lt;TableRow&gt;&gt;&gt; {
  
    @DefaultCoder(AvroCoder.class)
    public static class Accum implements Serializable {
      Double startTime = 0.0;
      Double endTime = 0.0;
    }
  
    @Override
    public Accum createAccumulator() {
      return new Accum();
    }
  
    @Override
    public Accum addInput(Accum accumulator, KV&lt;String, TableRow&gt; input) {
      // the earliest and latest times in the set of table rows is set on the accumulator
  
      return accumulator;
    }
  
    @Override
    public Accum mergeAccumulators(Iterable&lt;Accum&gt; accumulators) {
      Accum merged = createAccumulator();
      for (Accum accumulator : accumulators) {
        // merge steps happen here to find the earliest and latest times
      }
  
      return merged;
    }
  
    @Override
    public KV&lt;String, Iterable&lt;TableRow&gt;&gt; extractOutput(Accum accumulator) {
      // this step will create two rows based on the start and end times found in this function
    }
  }
  </code></pre>
  
  <p>I'm expecting the CombineFn to be compatible with Combine.GroupedValues as the documentation would seem to suggest. However, this is not the case. Combine.PerKey is another option, but we haven't found a way to use that as well with a CombineFn.</p>
  
  <p>Relevant links:<br>
  <a href="https://beam.apache.org/releases/javadoc/2.10.0/org/apache/beam/sdk/transforms/Combine.GroupedValues.html" rel="nofollow noreferrer">Documentation - Combine.GroupedValues</a><br>
  <a href="https://beam.apache.org/releases/javadoc/2.10.0/org/apache/beam/sdk/transforms/Combine.PerKey.html" rel="nofollow noreferrer">Documentation - Combine.PerKey</a>
  <a href="https://beam.apache.org/releases/javadoc/2.10.0/org/apache/beam/sdk/transforms/Combine.CombineFn.html" rel="nofollow noreferrer">Documentation - Combine.CombineFn</a><br>
  <a href="https://github.com/apache/beam/blob/release-2.10.0/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L214" rel="nofollow noreferrer">Source - Combine.GroupedValues</a><br>
  <a href="https://github.com/apache/beam/blob/release-2.10.0/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L165" rel="nofollow noreferrer">Source - Combine.PerKey</a>
  <a href="https://github.com/apache/beam/blob/release-2.10.0/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L330" rel="nofollow noreferrer">Source - Combine.CombineFn</a></p>