Created
April 1, 2019 12:03
-
-
Save gxercavins/199a91a49c90f9248d636f95e5b74e4b to your computer and use it in GitHub Desktop.
SO question 55219481: use session windows with Python SDK
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import argparse, json, logging, time | |
import apache_beam as beam | |
import apache_beam.transforms.window as window | |
from apache_beam.io import WriteToText | |
from apache_beam.options.pipeline_options import PipelineOptions | |
from apache_beam.options.pipeline_options import SetupOptions | |
class DebugPrinter(beam.DoFn): | |
"""Just prints the element with logging""" | |
def process(self, element, window=beam.DoFn.WindowParam): | |
for x in element[1]: | |
logging.info(">>> Received %s %s with window=%s", x['value'], x['timestamp'], window) | |
yield element | |
def run(argv=None): | |
parser = argparse.ArgumentParser() | |
known_args, pipeline_args = parser.parse_known_args(argv) | |
pipeline_options = PipelineOptions(pipeline_args) | |
pipeline_options.view_as(SetupOptions).save_main_session = True | |
p = beam.Pipeline(options=pipeline_options) | |
session_gap = 5 # [s]; 5 seconds | |
data = [{'user_id': 'Thanos', 'value': 'event_{}'.format(event), 'timestamp': time.time() + 2**event} for event in range(5)] | |
events = (p | |
| 'Create Events' >> beam.Create(data) \ | |
| 'Add Timestamps' >> beam.Map(lambda x: beam.window.TimestampedValue(x, x['timestamp'])) \ | |
| 'keyed_on_user_id' >> beam.Map(lambda x: (x['user_id'], x)) | |
| 'user_session_window' >> beam.WindowInto(window.Sessions(session_gap), | |
timestamp_combiner=window.TimestampCombiner.OUTPUT_AT_EOW) \ | |
| 'Group' >> beam.GroupByKey() | |
| 'debug_printer' >> beam.ParDo(DebugPrinter())) | |
result = p.run() | |
result.wait_until_finish() | |
if __name__ == '__main__': | |
logging.getLogger().setLevel(logging.INFO) | |
run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment