title: Slowness / Lag in beam streaming pipeline in group by key stage
body: <h3>Context</h3>
  
  <p>Hi all, I have been using <code>Apache Beam</code> pipelines to generate columnar DB to store in <code>GCS</code>, I have a datastream coming in from <code>Kafka</code> and have a window of 1m. </p>
  
  <p>I  want to transform all data of that 1m window into a columnar DB file (<code>ORC</code> in my case, can be <code>Parquet</code> or anything else), I have written a pipeline for this transformation.</p>
  
  <h3>Problem</h3>
  
  <p>I am experiencing general slowness. I suspect it could be due to the group by key transformation as I have only key. Is there really a need to do that? If not, what should be done instead? I read that combine isn't very useful for this as my pipeline isn't really aggregating the data but creating a merged file. What I exactly need is an iterable list of objects per window which  will be transformed to ORC files.</p>
  
  <h3>Pipeline Representation</h3>
  
  <blockquote>
    <p>input -> window -> group by key (only 1 key) -> pardo (to create DB) -> IO (to write to GCS)</p>
  </blockquote>
  
  <h3>What I have tried</h3>
  
  <p>I have tried using the profiler, scaling horizontally/vertically. Using the profiler I saw more than 50% of the time going into group by key operation. I do believe the problem is of hot keys but I am unable to find a solution on what should be done. When I removed the <code>group by key</code> operation, my pipeline keeps up with the <code>Kafka</code> lag (ie, it doesn't seem to be an issue at <code>Kafka</code> end).</p>
  
  <h3>Code Snippet</h3>
  
  <pre><code>p.apply("ReadLines", KafkaIO.&lt;Long, byte[]&gt;read().withBootstrapServers("myserver.com:9092")
      .withTopic(options.getInputTopic())
      .withTimestampPolicyFactory(MyTimePolicy.myTimestampPolicyFactory())
      .withConsumerConfigUpdates(Map.of("group.id", "mygroup-id")).commitOffsetsInFinalize()
      .withKeyDeserializer(LongDeserializer.class)
      .withValueDeserializer(ByteArrayDeserializer.class).withoutMetadata())
      .apply("UncompressSnappy", ParDo.of(new UncompressSnappy()))
      .apply("DecodeProto", ParDo.of(new DecodePromProto()))
      .apply("MapTSSample", ParDo.of(new MapTSSample()))
      .apply(Window.&lt;TSSample&gt;into(FixedWindows.of(Duration.standardMinutes(1)))
          .withTimestampCombiner(TimestampCombiner.END_OF_WINDOW))
      .apply(WithKeys.&lt;Integer, TSSample&gt;of(1))
      .apply(GroupByKey.&lt;Integer, TSSample&gt;create())
      .apply("CreateTSORC", ParDo.of(new CreateTSORC()))
      .apply(new WriteOneFilePerWindow(options.getOutput(), 1));
  </code></pre>
  
  <h3>Wall Time Profile</h3>
  
  <p><a href="https://gist.github.com/anandsinghkunwar/4cc26f7e3da7473af66ce9a142a74c35" rel="nofollow noreferrer">https://gist.github.com/anandsinghkunwar/4cc26f7e3da7473af66ce9a142a74c35</a></p>