def expand(self, pcoll):
    return (((((pcoll | ('Window into Fixed Intervals' >> beam.WindowInto(window.FixedWindows(self.window_size)))) | ('Add timestamps to messages' >> beam.ParDo(AddTimestamps()))) | ('Add Dummy Key' >> beam.Map((lambda elem: (None, elem))))) | ('Groupby' >> beam.GroupByKey())) | ('Abandon Dummy Key' >> beam.MapTuple((lambda _, val: val))))
def expand(self, pcoll):
    return (((((pcoll | ('Window into Fixed Intervals' >> beam.WindowInto(window.FixedWindows(self.window_size)))) | ('Add timestamps to messages' >> beam.ParDo(AddTimestamps()))) | ('Add Dummy Key' >> beam.Map((lambda elem: (None, elem))))) | ('Groupby' >> beam.GroupByKey())) | ('Abandon Dummy Key' >> beam.MapTuple((lambda _, val: val))))