import udf # create a general Python UDF @udf(result_type=DataTypes.BIGINT()) def plus_one(i): return i + 1 table.select(plus_one(col('id'))).to_pandas() 1.1. Getting Started 17 pyflink-docs, Release release-1 @udf(result_type=DataTypes.BIGINT(), func_type='pandas') def pandas_plus_one(series): return series + 1 table.select(pandas_plus_one(col('id'))).to_pandas() /Users/duanchen/sourcecode/flink/flink-python/dev/ use the Python function in SQL API table_env.create_temporary_function("plus_one", plus_one) table_env.sql_query("SELECT plus_one(id) FROM {}".format(table)).to_pandas() Another example is UDFs used

import udf # create a general Python UDF @udf(result_type=DataTypes.BIGINT()) def plus_one(i): return i + 1 table.select(plus_one(col('id'))).to_pandas() 1.1. Getting Started 17 pyflink-docs, Release release-1 @udf(result_type=DataTypes.BIGINT(), func_type='pandas') def pandas_plus_one(series): return series + 1 table.select(pandas_plus_one(col('id'))).to_pandas() /Users/duanchen/sourcecode/flink/flink-python/dev/ use the Python function in SQL API table_env.create_temporary_function("plus_one", plus_one) table_env.sql_query("SELECT plus_one(id) FROM {}".format(table)).to_pandas() Another example is UDFs used