full_pipeline_1.py

with apache_beam.Pipeline(options=options) as p:
    rows = (
        p |
        ReadFromText("input.txt") |
        apache_beam.ParDo(Split())
    )

    timings = (
        rows |
        apache_beam.ParDo(CollectTimings()) |
        "Grouping timings" >> apache_beam.GroupByKey() |
        "Calculating average" >> apache_beam.CombineValues(
            apache_beam.combiners.MeanCombineFn()
        )
    )

    users = (
        rows |
        apache_beam.ParDo(CollectUsers()) |
        "Grouping users" >> apache_beam.GroupByKey() |
        "Counting users" >> apache_beam.CombineValues(
            apache_beam.combiners.CountCombineFn()
        )
    )

    to_be_joined = (
        {
            'timings': timings,
            'users': users
        } |
        apache_beam.CoGroupByKey() |
        apache_beam.ParDo(WriteToCSV()) |
        WriteToText("output.txt")
    )