def perform_tumbling_window_aggregation()

in pyflink-examples/StreamingFileSink/streaming-file-sink.py [0:0]

• 10 lines of code
• 1 McCabe index (conditional complexity)

def perform_tumbling_window_aggregation(input_table_name):
    # use SQL Table in the Table API
    input_table = table_env.from_path(input_table_name)

    tumbling_window_table = (
        input_table.window(
            Tumble.over("10.seconds").on("event_time").alias("ten_second_window")
        )
        .group_by("ticker, ten_second_window")
        .select("ticker, price.sum as price, ten_second_window.end as event_time")
    )

    return tumbling_window_table