Dataflow: using beam.combiners on the results of a previous beam.combiners

user_7762016 0
Viewed 308

I am using a Beam pipeline to count the frequencies of phone numbers for streaming data. I'm using sliding windows which repeat every 5 mins with total period of 15 mins, so as expected, for some inputs I am getting multiple outputs when the input falls in multiple windows.

After calculating the number of occurrences, I would like to find the mean value for the input feature. The inputs are tuples like:

('phone_number', '123')
('phone_number', '456')
('phone_number', '456')
('phone_number', '456')

The first part of pipeline is to count the frequency of each number:

| 'window' >> beam.WindowInto(window.SlidingWindows(900, 300))
| 'pair_with_one' >> beam.Map(lambda x: (x, 1))
| 'count_occurences' >> beam.combiners.Count.PerKey()

This correctly my calculation works correctly and I can count the frequency of each number, getting 3 results as there are 3 sliding windows in each period (in our case 2 of the 456 calls were in the same window and the third was in a different one):

(('phone_number', '123'), 1)
(('phone_number', '123'), 1)
(('phone_number', '123'), 1)
(('phone_number', '456'), 2)
(('phone_number', '456'), 2)
(('phone_number', '456'), 2)
(('phone_number', '456'), 1)
(('phone_number', '456'), 1)
(('phone_number', '456'), 1)

Now, I'd like to find the mean for each phone number over all the window values calculated, i.e.:

(('phone_number', '123'), 1.0)
(('phone_number', '456'), 1.5)

The next step in my pipeline is

| 'Find Means' >> beam.combiners.Mean.PerKey()

But this just gives me:

(('phone_number', '123'), 1.0)
(('phone_number', '123'), 1.0)
(('phone_number', '123'), 1.0)
(('phone_number', '456'), 2.0)
(('phone_number', '456'), 2.0)
(('phone_number', '456'), 2.0)
(('phone_number', '456'), 1.0)
(('phone_number', '456'), 1.0)
(('phone_number', '456'), 1.0)

Is there any way of doing another beam.combiners calculation on the results of a previous one?

apache-beam google-cloud-dataflow python
1 Answers

The reason beam.combiners.Mean.PerKey() gives you an incorrect output, is that the combiner gives you a single value calculated for each key+window.

However, there is more here. The reason for windowing in streaming processing is to ensure that the input is bounded before producing a result. That is, normally input to streaming pipelines are unbounded, meaning they never stop receiving data unless the pipeline is terminated. So its not possible to compute a value across all of the windows, as you would need to wait forever.

To me it looks like you are trying to calculate "The mean number of occurrences of a phone number in a 15 minute window, when comparing all possible sliding 15 minute windows, by sliding it ever 5 minutes". If this is not the case, please clarify to help me understand

Since we need to bound the calculation somehow, it may be possible to output a result periodically, i.e. for every window, and keep outputting a new result, updating it until the pipeline ends. This should be possible with StatefulDoFn.

For this I recommend:

  • Output the counts from your sliding windows into GlobalWindows
  • Store the sum and count to compute a mean in a StatefulDoFn
  • Output the computed mean periodically, or on every element and update the result downstream (i.e. overwriting the same row in BigQuery, or removing the extra rows when examining the BigQuery table with SQL)

Something like this:

class ComputeMeanStatefulDoFn(DoFn):
  TOTAL_STATE = CombiningStateSpec('total', sum)
  COUNT_STATE = CombiningStateSpec('count', sum)

  def process(self, element,
      total=DoFn.StateParam(TOTAL_STATE),
      count=DoFn.StateParam(COUNT_STATE)):
    key_phone_number, value_window_count = element
    current_count = count.read() + 1
    current_total = total.read() + value_window_count
    mean = current_total / current_count
    # You can emit every N results to reduce the volume
    # but please make sure to at least emit the first M << N results
    yield (key_phone_number, mean)
    total.add(value_window_count)
    count.add(1)

| 'window' >> beam.WindowInto(window.SlidingWindows(900, 300))
| 'pair_with_one' >> beam.Map(lambda x: (x, 1))
| 'count_occurences' >> beam.combiners.Count.PerKey()
| 'window_globally' >> beam.WindowInto(window.GlobalWindows)
| 'compute_mean_across_windows' >> beam.ParDo(ComputeMeanStatefulDoFn)

Essentially what's happening here is that the sum and count are stored to persistance/disk and we recompute a new mean every time a new element arrives in the global window.

Note: You will need to deal with emitting the updated mean for the same key multiple times. I.e. you may wish to overwrite a row in a BigQuery table containing your results.

Note: Depending on the semantics what you are trying to compute as well, you may want to emit empty windows from the SlidingWindows function so they are included in the downstream mean calculation.

Note: You cannot use a Combine.globally here as this would never terminate, due to the nature of unbounded input in a streaming pipeline. I believe this may error out if you try to launch a pipeline like that.

user_52655890
Related