code: def pandas_udaf():
      env = StreamExecutionEnvironment.get_execution_environment()
      env.set_parallelism(1)
      t_env = StreamTableEnvironment.create(
          stream_execution_environment=env
      )
  
      # define the source with watermark definition
      ds = env.from_collection(
          collection=[
              (Instant.of_epoch_milli(1000), "Alice", 110.1),
              (Instant.of_epoch_milli(4000), "Bob", 30.2),
              (Instant.of_epoch_milli(3000), "Alice", 20.0),
              (Instant.of_epoch_milli(2000), "Bob", 53.1),
              (Instant.of_epoch_milli(5000), "Alice", 13.1),
              (Instant.of_epoch_milli(3000), "Bob", 3.1),
              (Instant.of_epoch_milli(7000), "Bob", 16.1),
              (Instant.of_epoch_milli(10000), "Alice", 20.1),
          ],
          type_info=Types.ROW(
              [Types.INSTANT(), Types.STRING(), Types.FLOAT()]
          ),
      )
  
      table = t_env.from_data_stream(
          ds,
          Schema.new_builder()
          .column_by_expression("ts", "CAST(f0 AS TIMESTAMP_LTZ(3))")
          .column("f1", DataTypes.STRING())
          .column("f2", DataTypes.FLOAT())
          .watermark("ts", "ts - INTERVAL '3' SECOND")
          .build(),
      ).alias("ts", "name", "price")
  
      # define the sink
      t_env.create_temporary_table(
          "sink",
          TableDescriptor.for_connector("print")
          .schema(
              Schema.new_builder()
              .column("name", DataTypes.STRING())
              .column("total_price", DataTypes.FLOAT())
              .column("w_start", DataTypes.TIMESTAMP_LTZ())
              .column("w_end", DataTypes.TIMESTAMP_LTZ())
              .build()
          )
          .build(),
      )
  
      @udaf(result_type=DataTypes.FLOAT(), func_type="pandas")
      def mean_udaf(v):
          return v.mean()
  
      # define the tumble window operation
      table = (
          table.window(
              Tumble.over(lit(5).seconds).on(col("ts")).alias("w")
          )
          .group_by(col("name"), col("w"))
          .select(
              col("name"),
              mean_udaf(col("price")),
              col("w").start,
              col("w").end,
          )
      )
  
      # submit for execution
      table.execute_insert("sink").wait()
  
file_path: flink/flink-python/pyflink/examples/table/pandas/pandas_udaf.py
imports: import logging
  import sys
  from pyflink.common.time import Instant
  from pyflink.common import Types
  from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.table import DataTypes, TableDescriptor, Schema, StreamTableEnvironment
  from pyflink.table.expressions import lit, col
  from pyflink.table.udf import udaf
  from pyflink.table.window import Tumble
label_desc: Use image node_id 1 for calling a global function with example usage: pandas_udaf() and returns: v