def perform_sliding_window_aggregation()

in pyflink-examples/SlidingWindows/sliding-windows.py [0:0]

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

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

    sliding_window_table = (
        input_table.window(
            Slide.over("10.seconds")
            .every("5.seconds")
            .on("event_time")
            .alias("ten_second_window")
        )
        .group_by("ticker, ten_second_window")
        .select("ticker, price.min as price, ten_second_window.end as event_time")
    )

    return sliding_window_table