<a id="cmf-pyflink-package"></a>

# Package a PyFlink job for Confluent Manager for Apache Flink

Flink jobs are deployed in Confluent Platform with Confluent Manager for Apache Flink® (CMF), which is a
central management component of Confluent Platform for Apache Flink.
This topic walks you through configuring your PyFlink project for packaging with CMF.

## Prerequisites

Before you package a PyFlink job, you must meet the following prerequisites:

- Confluent Manager for Apache Flink installed using Helm. For installation instructions, see [Install Confluent Manager for Apache Flink with Helm](../../installation/helm.md#install-cmf-helm).
- A PyFlink project with Python dependencies managed by a package manager such as `pip`.
- Docker installed.

## Set up the project configuration

Follow these steps to set up your PyFlink project configuration.

1. Create a Python file for your PyFlink application. Following is an example `python_demo.py`:
   ```python
   import logging
   import sys

   from pyflink.table import TableEnvironment, EnvironmentSettings


   def python_demo():
       t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
       t_env.execute_sql(
           """
       CREATE TABLE orders (
         order_number BIGINT,
         price        DECIMAL(32,2),
         buyer        ROW<first_name STRING, last_name STRING>,
         order_time   TIMESTAMP(3)
       ) WITH (
         'connector' = 'datagen'
       )"""
       )

       t_env.execute_sql(
           """
           CREATE TABLE print_table WITH ('connector' = 'print')
             LIKE orders"""
       )
       t_env.execute_sql(
           """
           INSERT INTO print_table SELECT * FROM orders"""
       )


   if __name__ == "__main__":
       logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
       python_demo()
   ```
2. Create a Python dependencies file. You can create a `requirements.txt` file or use a modern Python package manager like `uv` to manage
   your PyFlink dependencies:
   ```bash
   cp-pyflink>2.0
   ```

## Deploy your PyFlink application with CMF

After you have developed your PyFlink application locally, you need to package it for deployment.
Following are the recommended packaging options.

### Package with a custom docker image

You should package the PyFlink job with a custom Docker image that contains the Python environment and
dependencies. You must already have the infrastructure in place for building Docker images
in a build pipeline. You base the custom Docker image on the `confluentinc/cp-flink:2.0.0-cp1`
image found on Docker Hub.

1. Create the Dockerfile. Following is an example Dockerfile:
   ```bash
   FROM debian:latest AS builder

   # Install dependencies first
   RUN set -ex; \
     apt-get update; \
     apt-get -y install gcc default-jdk; \
     rm -rf /var/lib/apt/lists/*

   # Set the correct JAVA_HOME
   ENV JAVA_HOME=/usr/lib/jvm/default-java

   # Setup python environment with uv
   COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/

   # Install python in the app folder so we can port it over
   ENV UV_PYTHON_INSTALL_DIR=/opt/flink/pyflink/.uv

   # Create the virtual environment
   RUN mkdir -p /opt/flink/pyflink
   WORKDIR /opt/flink/pyflink
   RUN uv venv --python 3.11 .venv

   # Copy the project and install dependencies. This demo only requires cp-pyflink,
   # but you can install dependencies the way you prefer. We suggest building a project
   # that can be setup by running `uv sync`.
   # Make sure to install the correct version of cp-pyflink
   COPY python_demo.py ./
   RUN uv pip install cp-pyflink>2.0

   # Build the final image, copy the python project and virtualenv in /opt/flink/pyflink
   FROM confluentinc/cp-flink:2.0.0-cp1

   COPY --from=builder --chown=flink:flink /opt/flink/pyflink/ /opt/flink/pyflink/
   ```
2. Build the docker image. The following shows an example command to do this:
   ```bash
   docker build -t pyflink-test:latest .
   ```
3. Define the Flink application like shown in the following example:
   ```yaml
   apiVersion: cmf.confluent.io/v1
   kind: FlinkApplication
   metadata:
     name: python-example
   spec:
     image: pyflink-test:latest
     flinkVersion: v2_0
     flinkConfiguration:
       metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
       metrics.reporter.prom.port: 9249-9250
       taskmanager.numberOfTaskSlots: "1"
     serviceAccount: flink
     jobManager:
       resource:
         cpu: 1
         memory: 1024m
     taskManager:
       resource:
         cpu: 1
         memory: 1024m
     job:
       jarURI: local:///opt/flink/opt/flink-python-2.0.0-cp1.jar
       state: running
       parallelism: 1
       upgradeMode: stateless
       entryClass: org.apache.flink.client.python.PythonDriver
       args:
         - -pyclientexec
         - /opt/flink/pyflink/.venv/bin/python3
         - -py
         - /opt/flink/pyflink/python_demo.py
   ```

### Submit the application definition

To run your PyFlink application with CMF you need to make the Python environment and dependencies available to the Flink clusters.
After you package the application, you can use the Confluent CLI to submit your Flink application definition.

For example:

```none
confluent flink application create --environment <env-name> <application-definition>
```

### Add Java dependencies

PyFlink applications often require Java JARs at runtime, for example, connectors,
formats, or custom Java UDFs. Because PyFlink does not use Maven or Gradle directly,
you must obtain the JARs separately and make them available in the container.

You can use a Maven `pom.xml` to declare the required Java dependencies and download
them in a single step. Use the Confluent Platform for Apache Flink Maven repository and the `io.confluent.flink`
group ID for supported components. For a list of supported components, see [Confluent Platform for Apache Flink Features and Support](supported-features.md#cpflink-vs-oss).

1. Create a `pom.xml` that contains the required Java dependencies for your PyFlink application:
   ```xml
   <?xml version="1.0" encoding="UTF-8"?>
   <project>
       <modelVersion>4.0.0</modelVersion>
       <groupId>com.example</groupId>
       <artifactId>pyflink-dependencies</artifactId>
       <version>1.0</version>
       <packaging>pom</packaging>
       <repositories>
           <repository>
               <id>cp-flink-releases</id>
               <url>https://packages.confluent.io/maven</url>
               <releases>
                   <enabled>true</enabled>
               </releases>
               <snapshots>
                   <enabled>false</enabled>
               </snapshots>
           </repository>
       </repositories>
       <dependencies>
           <dependency>
               <groupId>io.confluent.flink</groupId>
               <artifactId>flink-sql-connector-kafka</artifactId>
               <version>4.0.1-2.0-cp1</version>
           </dependency>
           <!-- Add other dependencies as needed (formats, UDFs, etc.) -->
       </dependencies>
   </project>
   ```
2. Download the JARs locally using Maven:
   ```bash
   mvn dependency:copy-dependencies -DoutputDirectory=./jars
   ```

   This copies all required JARs into the `./jars` directory.
3. Add the JARs to your Docker image by extending the Dockerfile. Add the following line
   after the `COPY --from=builder` step:
   ```bash
   COPY --chown=flink:flink jars/ /opt/flink/lib/
   ```

   JARs placed in `/opt/flink/lib/` are automatically available on the classpath and do not
   require additional `pipeline.jars` configuration.

## Important notes

Note the following about deploying PyFlink applications with CMF:

- Python Version Compatibility: Ensure your Python version is compatible with the PyFlink version you’re using. See [Versions and Interoperability for Confluent Manager for Apache Flink](../../installation/versions-interoperability.md#cmf-interop) for more information.
- Flink Python JAR: The `jarURI` should point to the Flink Python JAR file that
  matches your Flink version. The path is typically
  `/opt/flink/opt/flink-python-<version>.jar`.
- Entry Class: PyFlink applications use `org.apache.flink.client.python.PythonDriver` as the entry class.
- Python Executor: The `-pyclientexec` argument should point to the Python executable in your container or its virtual environment.
- Python Script Path: The `-py` argument should point to the path of your Python script within the container.
- Resource Requirements: Adjust the CPU core and memory requirements based on your application’s needs.
- Dependencies: Make sure all required Python packages are installed in your container.

## Troubleshooting

Following are some troubleshooting tips if you encounter issues when deploying PyFlink applications
with CMF:

- Python Environment Issues: Ensure the Python virtual environment is properly set up and all dependencies are installed.
- JAR File Path: Verify that the Flink Python JAR file exists at the specified path in your container.
- Permissions: Make sure the `flink` user has read access to the Python script and dependencies.
- Logs: Check the Flink job manager and task manager logs for Python-related errors.

## Related content

- [Flink Jobs for Confluent Manager for Apache Flink](../overview.md#af-jobs-cmf)