title: How to transform bounded pcollection to unbounded in Python with Apache Beam?
body: <p>I'm trying to transform a few Terabytes of mail logs stored in GCS without using too much memory.  </p>
  
  <p>As recommended in the <a href="https://beam.apache.org/documentation/programming-guide/#adding-timestamps-to-a-pcollections-elements" rel="nofollow noreferrer">guide</a>, I'm adding timestamp to each element, split it to a sliding window and I have specified an (aggregation) trigger before sending it to a <code>GroupByKey</code> and a <code>ParDo</code> parser after. This should do, but still <code>GroupByKey</code> waits for all data to arrive. <strong>Why?</strong></p>
  
  <p>I have tried it with Direct and Google Dataflow runner as well.</p>
  
  <p>What do I miss?</p>
  
  <p>Here's the gist of the code:</p>
  
  <pre class="lang-python prettyprint-override"><code>    known_args, pipeline_options = parse_args(sys.argv)
      pipeline = beam.Pipeline(options=pipeline_options)
      lines = pipeline | 'read' &gt;&gt; beam.io.ReadFromText(known_args.input)
      parsed_json = lines \
        | 'cut' &gt;&gt; beam.Map(lambda x: x[39:])\
        | 'jsonify' &gt;&gt; beam.Map(json_loads).with_outputs('invalid_input', main='main')
  
      keyed_lines = parsed_json['main']\
        | 'key_filter' &gt;&gt; beam.Filter(lambda x: u'key' in x)\
        | 'keyer' &gt;&gt; beam.Map(lambda x: (x['key'], x))\
        | 'log_processed_rows' &gt;&gt; beam.Map(log_processed_rows)
  
      window_trigger = trigger.DefaultTrigger()
  
      windowed_lines = keyed_lines\
         | 'timestamp' &gt;&gt; beam.ParDo(AddTimestampDoFn())\
         | 'window' &gt;&gt; beam.WindowInto(beam.window.SlidingWindows(size=3600+600, period=3600), trigger=window_trigger, 
                                       accumulation_mode=trigger.AccumulationMode.DISCARDING) # 1 hour steps with 10 minutes of overlap
  
      results = windowed_lines \
          | 'groupbykey' &gt;&gt; beam.GroupByKey()\
          | 'parse' &gt;&gt; beam.ParDo(ParseSendingsDoFn()).with_outputs('too_few_rows', 'invalid_rows', 'missing_recipients', main='main_valid')  
  
      output = results['main_valid'] \
          | 'format' &gt;&gt; beam.Map(output_format)\
          | 'write' &gt;&gt; beam.io.WriteToText(known_args.output, file_name_suffix=".gz")
  </code></pre>