title: How to combine streaming data with large history data set in Dataflow/Beam
body: <p>I am investigating processing logs from web user sessions via Google Dataflow/Apache Beam and need to combine the user's logs as they come in (streaming) with the history of a user's session from the last month.</p>
  
  <p>I have looked at the following approaches:</p>
  
  <ol>
  <li>Use a 30 day fixed window: most likely to large of a window to fit into memory, and I do not need to update the user's history, just refer to it</li>
  <li>Use CoGroupByKey to join two data sets, but the two data sets must have the same window size (<a href="https://cloud.google.com/dataflow/model/group-by-key#join">https://cloud.google.com/dataflow/model/group-by-key#join</a>), which isn't true in my case (24h vs 30 days)</li>
  <li>Use Side Input to retrieve the user's session history for a given <code>element</code> in <code>processElement(ProcessContext processContext)</code></li>
  </ol>
  
  <p>My understanding is that the data loaded via <code>.withSideInputs(pCollectionView)</code> needs to fit into memory. I know I can fit all of a single user's session history into memory, but not <em>all</em> session histories. </p>
  
  <p>My question is if there is a way to load/stream data from a side input that is only relevant to the current user session?</p>
  
  <p>I am imagining a parDo function that would load the user's history session from the side input by specifying the user's ID. But only the current user's history session would fit into memory; loading <em>all</em> history sessions through side input would be too large.</p>
  
  <p>Some pseudo code to illustrate:</p>
  
  <pre><code>public static class MetricFn extends DoFn&lt;LogLine, String&gt; {
  
      final PCollectionView&lt;Map&lt;String, Iterable&lt;LogLine&gt;&gt;&gt; pHistoryView;
  
      public MetricFn(PCollectionView&lt;Map&lt;String, Iterable&lt;LogLine&gt;&gt;&gt; historyView) {
          this.pHistoryView = historyView;
      }
  
      @Override
      public void processElement(ProcessContext processContext) throws Exception {
          Map&lt;String, Iterable&lt;LogLine&gt;&gt; historyLogData = processContext.sideInput(pHistoryView);
  
          final LogLine currentLogLine = processContext.element();
          final Iterable&lt;LogLine&gt; userHistory = historyLogData.get(currentLogLine.getUserId());
          final String outputMetric = calculateMetricWithUserHistory(currentLogLine, userHistory);
          processContext.output(outputMetric);
      }
  }
  </code></pre>