peak_pricing_table = table_env.from_data_stream(peak_pricing_events_stream,Schema.new_builder().build()) peak_pricing_table.print_schema()

(
  `timestamp` TIMESTAMP(9),
  `dt` STRING,
  `hr` INT,
  `mnt` INT,
  `5_mnt_btch` INT,
  `user_id` STRING,
  `pickupgeohash` STRING,
  `dropgeohash` STRING,
  `city` STRING,
  `category` STRING,
  `device` STRING,
  `ride_distance` DOUBLE,
  `AR` DOUBLE
)

table_env.create_temporary_view("sql_full_df",peak_pricing_table )

table_env.execute_sql("""
    SELECT 
        `timestamp`, window_start, window_end
    FROM TABLE(TUMBLE(TABLE sql_full_df, DESCRIPTOR(`timestamp`), INTERVAL '5' MINUTES))
""").print()