Apache beam-fixed窗口，具有产生早期结果的默认触发器

Question

null

Counts, Unix_Timestamp
1, 1553578200
2, 1553578201
3, 1553578202
4, 1553578203
...

def encode_byte_string(element):
  #element = ', '.join(element)
  #count = str(count)
  element = str(element)
  print element
  return element.encode('utf-8')

def custom_timestamp(message):
  data, time_stamp = message.split(',')
  # assuming that message is already parsed JSON (dict)
  return beam.window.TimestampedValue(data, int(time_stamp))

class BuildRecordFn(beam.DoFn):
  def process(self, element, window=beam.DoFn.WindowParam):
    window_start = window.start
    window_end = window.end
    return [element + (window_start,) + (window_end,)]


pubsub_data = (
                p 
                | 'Read from pub sub' >> beam.io.ReadFromPubSub(subscription= input_subscription)
                | 'Remove extra chars' >> beam.Map(lambda data: (data.rstrip().lstrip()))
                | 'CustomTimestamp' >> beam.Map(custom_timestamp)
                | 'Window' >> beam.WindowInto(window.FixedWindows(100))
                | 'Form Key Value pair' >> beam.Map(lambda x: (1, int(x)))
                | 'Sum values' >> beam.GroupByKey()
                | 'AddWindowEndTimestamp' >> beam.ParDo(BuildRecordFn())
                | 'Encode to byte string' >> beam.Map(encode_byte_string)
                | 'Write to pub sub' >> beam.io.WriteToPubSub(output_topic)
              )


result = p.run()
result.wait_until_finish()

(1, [1,2,3,4, ..., 99], Timestamp(1553578200), Timestamp(1553578300))
(1, [99,100,101,102, ..., 199], Timestamp(1553578300), Timestamp(1553578400))

然而，当我运行这个管道时，我甚至在通过窗口结束之前就得到了早期结果

(1, [11], Timestamp(1553578200), Timestamp(1553578300))
(1, [12, 16, 15], Timestamp(1553578200), Timestamp(1553578300))
(1, [19, 18, 8, 10, 23, 21, 1, 7, 9, 13], Timestamp(1553578200), Timestamp(1553578300))
(1, [5, 6, 14, 17, 20, 22], Timestamp(1553578200), Timestamp(1553578300))
(1, [33], Timestamp(1553578200), Timestamp(1553578300))
(1, [3], Timestamp(1553578200), Timestamp(1553578300))
(1, [24, 28, 29, 37, 39, 43], Timestamp(1553578200), Timestamp(1553578300))
(1, [2, 4], Timestamp(1553578200), Timestamp(1553578300))
(1, [48], Timestamp(1553578200), Timestamp(1553578300))
(1, [25, 31, 34, 36, 38, 40, 46, 49, 51], Timestamp(1553578200), Timestamp(1553578300))
(1, [26, 27, 30, 32, 41, 42, 45, 47], Timestamp(1553578200), Timestamp(1553578300))
(1, [44, 52], Timestamp(1553578200), Timestamp(1553578300))
(1, [35, 50], Timestamp(1553578200), Timestamp(1553578300))

这可能是什么原因？

Answer 1

似乎是这里描述的晚期数据的情况。https://beam.apache.org/documentation/programming-guide/#水印和后期数据

缓解此类问题的一种方法是使用WithAllowedlateness添加某种“延迟”。但是，这个函数现在似乎只在Java SDK中可用。

另一种方法可能是使用afterwatermark触发器。但是，我还没试过。所以，带着点盐去吧。