Realtime Compute for Apache Flink:Develop PyFlink jobs

Prerequisites

Before you begin, make sure you have:

• A workspace in Realtime Compute for Apache Flink

• An IDE such as PyCharm or VS Code

Constraints and limits

Set up your local environment

python --version

pip install apache-flink==1.17.2

Choose an API

PyFlink supports Table API/SQL and DataStream API. Table API/SQL is strongly recommended for most use cases:

Use DataStream API only when SQL cannot express your complex custom logic.

Set up your project

The recommended project structure:

my-flink-python-project/
├── my_job.py                # Main job file
├── udfs.py                  # User-defined functions (optional)
├── requirements.txt         # Third-party Python dependencies (optional)
└── config.properties        # Configuration file (optional)

Manage dependencies

To use custom Python virtual environments, third-party Python packages, JARs, or data files in PyFlink jobs, see Use Python dependencies.

Write user-defined functions

The following example shows a user-defined scalar function (UDSF) that masks sensitive data in phone numbers:

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())
def mask_phone(phone: str):
    """Mask phone number: keep the first three and last four digits; replace middle digits with ****."""
    if phone is None or len(phone) != 11:
        return phone
    return phone[:3] + '****' + phone[7:]

Register the UDF in a SQL job:

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;
INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

To register, update, or delete UDFs, see Manage user-defined functions (UDFs).

Configure connectors

Connectors let your PyFlink job read from and write to external systems. For supported connectors, see Supported connectors.

1. Log on to the Realtime Compute for Apache Flink console.

2. Click Console in the Actions column of the target workspace.

3. In the left navigation pane, click Artifacts.

4. Click Upload Artifact, and select the Python package for the target connector. Download official Python packages for Flink connectors from the connector list.

5. On the O\&M \> Deployments page, click Create Deployment \> Python Job, select the connector package in Additional Dependencies, configure the other parameters, and deploy the job.

6. Click the job name. On the Configuration tab, go to Parameters, click Edit, and add the connector package path in Other Configuration. For multiple connector JARs, separate them with semicolons:

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. Optional: To use built-in connectors, formats, or catalogs (available in VVR 11.2 and later), add the following configuration in Other Configuration under Parameters:

   ## Use multiple built-in connectors
   pipeline.used-builtin-connectors: kafka;sls
   ## Use multiple data formats
   pipeline.used-builtin-formats: avro;parquet
   ## Use multiple existing catalogs
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

For complete connector usage examples, see Complete sample code.

Debug a job

When you implement a Python UDF, use the logging module to output log information for troubleshooting:

import logging

@udf(result_type=DataTypes.BIGINT())
def add(i, j):
    logging.info("Computing add: %s + %s", i, j)
    return i + j

View the output in the TaskManager log files.

Deploy a job

After you finish developing your PyFlink job, deploy it to Realtime Compute for Apache Flink:

1. Log on to the Realtime Compute for Apache Flink console and go to the target workspace.

2. In the left navigation pane, click Artifacts. Upload your Python job files (.py or .zip), third-party dependencies, and configuration files.

3. On the O\&M \> Deployments page, click Create Deployment \> Python Job, and fill in the deployment parameters:

   Parameter	Description
   Python URI	Select the uploaded Python job file.
   Entry Module	Leave blank for .py files. For .zip files, enter the entry module name — for example, my_job.
   Additional Dependencies	Select connector JARs or configuration files, if any.
   Python Libraries	Select third-party Python packages (.whl or .zip), if any.
   Python Archives	Select custom Python virtual environments (.zip), if any.

4. Click Deploy.

For details about deployment parameters, see Deploy a job.

Complete sample code

The following example reads data from Kafka, filters records, and writes results to MySQL. All table definitions use the Table API and SQL.

This example does not include configurations for checkpoints or restart strategies. Add these after deployment on the Configuration tab. For details, see Configure job deployment settings.

import logging
import sys

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

logging.basicConfig(stream=sys.stdout, level=logging.INFO)


def kafka_to_mysql():
    # Create execution environments
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    # Create Kafka source table
    t_env.execute_sql("""
        CREATE TABLE kafka_source (
            `id` INT,
            `name` STRING,
            `score` INT,
            `event_time` TIMESTAMP(3),
            WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'student_topic',
            'properties.bootstrap.servers' = 'your-kafka-broker:9092',
            'properties.group.id' = 'my-group',
            'scan.startup.mode' = 'latest-offset',
            'format' = 'json'
        )
    """)

    # Create MySQL sink table
    t_env.execute_sql("""
        CREATE TABLE mysql_sink (
            `id` INT,
            `name` STRING,
            `score` INT,
            PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:mysql://your-mysql-host:3306/my_database',
            'table-name' = 'student',
            'username' = 'your_username',
            'password' = 'your_password'
        )
    """)

    # Filter records with score >= 60 and write to MySQL
    t_env.execute_sql("""
        INSERT INTO mysql_sink
        SELECT id, name, score
        FROM kafka_source
        WHERE score >= 60
    """)


if __name__ == '__main__':
    kafka_to_mysql()

Replace the following placeholders:

Pre-installed software packages

What's next

• Quick start with PyFlink jobs — end-to-end walkthrough of the PyFlink job development process

• Use Python dependencies — custom Python virtual environments, third-party libraries, JARs, and data files

• Job development overview and Develop JAR jobs — SQL and DataStream job development guides