--- title: "How to run PyFlink Jobs and Python UDFs on Ververica Platform" description: "Learn how to run PyFlink jobs and Python UDFs on Ververica Platform with a custom Docker image, from setup to execution in Java Table API programs." lastUpdated: 2026-10-14T09:41:47.000Z source_url: html: "https://www.ververica.com/developers/how-to-run-pyflink-jobs-and-python-udfs-on-ververica-platform" md: "https://www.ververica.com/developers/how-to-run-pyflink-jobs-and-python-udfs-on-ververica-platform.md" --- [PyFlink](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/overview/) is a Python API for Apache Flink. It provides a way to write Flink programs in Python and execute them on a Flink cluster. Some existing Ververica Platform users may be wondering whether they can run PyFlink jobs on Ververica Platform while other non-users may think that it is not possible to run their Flink jobs coded in Python on the platform. This tutorial will show you how you can run PyFlink jobs and Python User Defined Functions (UDFs) via a custom Flink Docker image with Python and PyFlink installed. We will show you two ways to run Python code in your Flink jobs on Ververica Platform: 1. run PyFlink jobs directly 1. call Python UDFs in your Table API programs written in Java If you would like to follow along, you can download the [full example code from this post](https://github.com/ververica/lab-vvp-pyflink). We will be using Ververica Platform 2.8, Flink 1.15.2, and Python 3.8 in our example. Please change accordingly if you use different versions. ## Prerequisites This tutorial assumes that you have the following: - A recent Ververica Platform installed and running with a [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html) configured. Refer to [the Ververica Platform Getting Started instructions](https://docs.ververica.com/getting_started/installation.html) if you do not have this set up yet. - Familiarity with the Flink job lifecycle management on Ververica Platform. This [short video](https://www.youtube.com/watch?v=BgXuxCZXNS0) will help you get started if you are not. If you are running Ververica Platform 2.8 or earlier, or Ververica Platform 2.9 with Flink 1.13 or 1.14, you will need: - A [docker runtime](https://docs.docker.com/engine/install/) to build Docker images - A container registry to push your Docker images to ## Create a custom Flink image that includes Python and PyFlink In order to run PyFlink jobs or call Python UDFs, you first need a Python interpreter and PyFlink. These are included in the Flink 1.15.3 (or later) images that come with Ververica Platform 2.9 (or later). You can skip this section and use the built-in Flink images if you are using these Ververica Platform versions. If you are using Ververica Platform 2.8 or earlier, or if you are using Flink 1.13/1.14 with Ververica Platform 2.9, follow these steps to build new Flink images that include a Python interpreter and PyFlink. In your terminal, create an empty directory. In the directory, create a file named Dockerfile with the following content: ```dockerfile FROM registry.ververica.com/v2.8/flink:1.15.2-stream4-scala_2.12-java11 USER root:root RUN apt update \ && apt -y install python3.8 python3-pip \ && python3.8 -m pip install apache-flink==1.15.2 \ && apt clean \ && ln -s /usr/bin/python3.8 /usr/bin/python \ && rm -rf /var/lib/apt/lists/* USER flink:flink ``` This Dockerfile uses a Flink 1.15.2 image (provided by Ververica Platform 2.8) as its base image. Build the docker image by using the `docker build` command: ```dockerfile docker build -t 1.15.2-stream4-scala_2.12-java11-python3.8-pyflink . ``` Once the image is built, you can tag and push the image to a public or private container registry that is reachable by Ververica Platform: ```dockerfile docker tag 1.15.2-stream4-scala_2.12-java11-python3.8-pyflink:latest /flink:1.15.2-stream4-scala_2.12-java11-python3.8-pyflink docker push /flink:1.15.2-stream4-scala_2.12-java11-python3.8-pyflink ``` ## Create a Python UDF In PyFlink, a Python UDF is a function written in Python that can be used in a Flink program. PyFlink provides a number of APIs for defining and using Python UDFs in Flink programs. Python UDFs are useful when you want to perform custom operations on your data that are not provided by the built-in functions in Flink, or when you want to use existing Python libraries in your Flink programs. Throughout this tutorial, you will use a simple Python UDF which takes a string as input and returns it in capitalized form. This UDF can [be found here](https://github.com/ververica/lab-vvp-pyflink/blob/main/python-udf/my_udfs.py) if you want to follow along. ```python from pyflink.table.udf import udf from pyflink.table import DataTypes @udf(result_type=DataTypes.STRING()) def py_upper( str ): "This capitalizes the whole string" return str.upper() ``` The annotation `@udf(result_type=DataTypes.STRING())` is the glue that is needed to make this function available as a Python UDF. For other ways to define Python UDFs, please consult the official [Flink documentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/udfs/python_udfs/#scalar-functions). ## Run PyFlink Jobs on Ververica Platform You will be running this [PyFlink job](https://github.com/ververica/lab-vvp-pyflink/blob/main/pyflink-job/pyflink_table_example.py) on Ververica Platform. This job relies on the aforementioned Python UDF to work. You will run it with Flink’s PythonDriver in [flink-python_2.12-1.15.2.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-python_2.12/1.15.2/flink-python_2.12-1.15.2.jar). This is also how Apache Flink runs Python pipelines internally. First, upload [pyflink_table_example.py](https://github.com/ververica/lab-vvp-pyflink/blob/main/pyflink-job/pyflink_table_example.py) and [my_udfs.py](https://github.com/ververica/lab-vvp-pyflink/blob/main/python-udf/my_udfs.py) to Ververica Platform’s [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html) via the menu bar on the left: **Deployments > Artifacts** ![Ververica Platform menu with Deployments > Artifacts selected, opening the Artifacts upload page](https://cdn.sanity.io/images/6b38iw1w/production/87ec49c6d4f800e96b492e0438202e3ca0bfb3dd-600x405.heif) Then create your deployment with the following values: The screenshot below is an example of how to do it on the Ververica Platform UI: ![Ververica Platform deployment configuration for PyFlink: flink-python JAR, Python dependencies, PythonDriver entry point](https://cdn.sanity.io/images/6b38iw1w/production/e53a55a0e8d1d1481071e1843f40bb1432440a17-950x700.heif) The directory `/flink/usrlib` is used in main args. When a Flink job starts, Ververica Platform will download the artifacts from its [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html) to `/flink/usrlib`. Now, you can start this deployment, which will run the PyFlink job. Since the job has a bounded data input, the deployment will eventually go to the FINISHED state with Flink pods being torn down. But you should be able to get the following result in the logs of the JobManager pod where the city column is capitalized as expected: ``` +----+----------------------+--------------------------------+ | op | id | city | +----+----------------------+--------------------------------+ | +I | 1 | BERLIN | | +I | 2 | SHANGHAI | | +I | 3 | NEW YORK | | +I | 4 | HANGZHOU | +----+----------------------+--------------------------------+ ``` ## Call Python UDFs in a Java Table API Job Now that you have seen how to run PyFlink jobs on Ververica Platform, you can try calling Python UDFs from a Table API program written in Java. This is useful when several teams collaborate on Python UDFs. For example, one team of domain experts may develop Python UDFs and provide them to another team of Java developers for use in their Table API jobs. You will use this [Table API job](https://github.com/ververica/lab-vvp-pyflink/blob/main/java-table-job/src/main/java/com/ververica/JavaTableExample.java) in this example, which also calls the aforementioned Python UDF `my_udfs.py`. Before you run the Table API job, let's go over the essential pieces that make this work. The first step is to add a job dependency. More specifically, youneed to add the dependency `flink-python` into our project and package it into the job jar. See the full [pom.xml](https://github.com/ververica/lab-vvp-pyflink/blob/main/java-table-job/pom.xml) file for details. The dependency is where Flink’s PythonDriver exists. ```xml org.apache.flink flink-python_2.12 1.15.2 compile ``` Then, in the Table API job [JavaTableExample](https://github.com/ververica/lab-vvp-pyflink/blob/main/java-table-job/src/main/java/com/ververica/JavaTableExample.java), you need to inform your job where it can look for any UDF files. You can do this via PyFlink’s configuration [python.files](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/python_config/#python-files). Remember, `/flink/usrlib` is the directory where Ververica Platform downloads the artifacts to when it starts a Flink job. ```python tableEnv.getConfig().getConfiguration() .setString("python.files", "/flink/usrlib/my_udfs.py"); ``` Next, call `executeSql()` to map the Python function `my_udfs.py_upper` (since the Python UDF is in a separate file, youneed to reference it with this format python_module.function_name) to a temporary system function `PY_UPPER` in Flink SQL. ```python tableEnv.executeSql("create temporary system function PY_UPPER as 'my_udfs.py_upper' language python"); ``` Now that `PY_UPPER` is registered in Flink SQL, you can use the function in your SQL statement. ```python tableEnv.sqlQuery( "SELECT user, PY_UPPER(product), amount FROM " + table) ``` Before you can run the job, you need to upload the Python UDF file `my_udfs.py` to Ververica Platform’s [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html). You can skip this if you have done so by following the steps in the earlier part of this blog post. Then, package the job as a jar: go to the directory with the `pom.xml` file, run `mvn clean package`, and you will get the JAR file `JavaTableExample-1.0-SNAPSHOT.jar` under the directory `target`. With the produced jar file, you can create a deployment as you normally do with any other jar-based deployments. The only two differences here are that you have to: 1. make sure the custom Flink image `1.15.2-stream4-scala_2.12-java11-python3.8-pyflink` (or the built-in one if it has Python and PyFlink included) is used 1. add the Python UDF file `my_udfs.py` as an additional dependency. Now start this deployment. The Python UDF will be called during the job run. Similar to the previous example, this job also has a bounded data input, so the deployment will eventually go to the FINISHED state with Flink pods being torn down. But you should be able to get the following result in the Jobmanager pod logs where the product column is capitalized as expected: ``` +----+----------------------+--------------------------------+-------------+ | op | user | product | amount | +----+----------------------+--------------------------------+-------------+ | +I | 1 | BEER | 3 | | +I | 2 | APPLE | 4 | +----+----------------------+--------------------------------+-------------+ ``` ## Review the steps Hopefully, you have successfully gone through this guide and have now seen how to run PyFlink jobs and Python UDFs on Ververica Platform. Let’s review the steps we just covered: 1. Create a custom Flink container image with Python and PyFlink installed 1. Run PyFlink jobs 1. Upload PyFlink jobs and its UDFs to Ververica Platform’s [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html) 1. Use Flink’s PythonDriver to run PyFlink Jobs 1. Reference the Python files in the main args. 1. Deploy with the Python files as additional dependencies. 1. Call Python UDFs in a Java based Table API program 1. Upload Python UDFs to Ververica Platform’s [Universal Blob Storage](https://docs.ververica.com/platform_operations/blob_storage.html) 1. Include `flink-python_2.12` as a dependency and package it into the job jar 1. Set `python.files` and register Python UDFs in the table environment 1. Deploy with the Python files as additional dependencies. ## Summary In this tutorial you created a custom Flink image with a Python interpreter and PyFlink in it. You then ran Python code in Flink jobs on Ververica Platform in two different ways: 1. run PyFlink jobs with Flink’s PythonDriver 1. call Python UDFs in a Java-based Table API program Ververica Platform 2.9 makes it easier by providing a Python interpreter and PyFlink in its Flink 1.15.3 images out of the box. As a next step, we are working on the native integration of PyFlink into Ververica Platform in order to further improve the overall PyFlink user experience. Stay tuned! ## Frequently Asked Questions ### What is PyFlink? PyFlink is a Python API for Apache Flink. It provides a way to write Flink programs in Python and execute them on a Flink cluster. ### What is a Python UDF in PyFlink and when should you use one? In PyFlink, a Python UDF is a function written in Python that can be used in a Flink program. Python UDFs are useful for custom operations on data that Flink's built-in functions do not provide, or when you want to use existing Python libraries in Flink programs. ### How do you run a PyFlink job on Ververica Platform? A PyFlink job on Ververica Platform runs with Flink's PythonDriver from the flink-python JAR, which is also how Apache Flink runs Python pipelines internally. The Python files are uploaded to Ververica Platform's Universal Blob Storage, referenced in the main args and added as additional dependencies, and the Flink image must include a Python interpreter and PyFlink. ### Can you call Python UDFs from a Java Table API job in Flink? Yes. You add the flink-python dependency to the job JAR, point the job to the UDF files with the python.files configuration, and register the Python function as a temporary system function with executeSql() so it can be used in Flink SQL statements. This is useful when one team develops Python UDFs and another team uses them in Java Table API jobs.