Realtime Compute for Apache Flink:Python Job Development

Choosing between Table API/SQL and DataStream API

PyFlink supports both Table API/SQL and DataStream API. We recommend using Table API/SQL first for these reasons:

• Better performance: Table API/SQL execution plans are optimized and run entirely within the JVM. DataStream API requires serialization and deserialization of each record between the JVM and Python processes, which incurs significant overhead.

• More complete features: Table API/SQL offers fuller support for connectors, data formats, and window functions, and shares the same connector ecosystem as SQL jobs.

• Community direction: The Apache Flink community focuses PyFlink development primarily on Table API/SQL.

Use DataStream API only for complex custom logic that cannot be expressed in SQL.

Development reference

Use the following documents to develop Flink business code locally. After development, upload your code to the Flink development console and deploy the job.

• For Apache Flink 1.20 business code development, see Flink Python API Development Guide.

• For common issues and solutions during Apache Flink coding, see FAQ.

Project structure

We recommend the following project structure for Python jobs:

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)

Dependency management

For instructions on using custom Python virtual environments, third-party Python packages, JARs, and data files in Python jobs, see Use Python dependencies.

User-defined functions (UDFs)

The following example shows how to develop a Python UDSF that desensitizes strings:

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

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

Use this UDF in a SQL job as follows:

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;

For instructions on registering, updating, and deleting UDFs, see Manage user-defined functions (UDFs).

Use connectors

For a list of supported connectors, see Supported connectors. To use a connector, follow these steps:

1. Log on to the Realtime Compute console.

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

3. In the left navigation pane, click File Management.

4. Click Upload Artifact and select the Python package for your target connector.

   You can upload a custom connector or a built-in Flink connector. For download links to official Flink connector Python packages, see Connector list.

5. On the O&M > Job O&M page, click Deploy Job > Python Job. In the Additional Dependencies field, select your connector package, configure other parameters, and deploy the job.

6. Click the deployed job name. On the Deployment Details tab, in the Runtime Parameters section, click Edit. In the Other Configuration field, add the path to your Python connector package.

   If your job depends on multiple connector packages—for example, connector-1.jar and connector-2.jar—configure them as follows:

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

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

   pipeline.used-builtin-connectors: kafka;sls
   pipeline.used-builtin-formats: avro;parquet

Job debugging

import logging

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

Local debugging

By default, Realtime Compute for Apache Flink cannot access the public network, so your code might not connect directly to online data sources for testing. Use one of these approaches for local debugging:

• Unit tests: Test user-defined functions (UDFs) independently to verify logic.

• Local execution: Simulate input using local data sources such as files or in-memory data, then run the job locally to validate processing logic. For example:

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()
  # Use a local data source for testing
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• Remote debugging: To debug with online data sources, see Run and debug connector-based jobs locally.

Job deployment

After developing a Python job, upload it to the Realtime Compute console for deployment. Follow these steps:

1. Log on to the Realtime Compute console and go to your target workspace.

2. In the navigation pane on the left, click File Management. Upload your Python job file (.py or .zip), along with any third-party dependencies or configuration files.

3. On the O&M > Job O&M page, click Deploy Job > Python Job and enter deployment information.

   Parameter	Description
   Python file path	Select the uploaded Python job file.
   Entry Module	Leave blank if your job file is a .py file. If it is a .zip file, enter the entry module name, such as my_job.
   Additional Dependencies	Select any connector JARs or configuration files here.
   Python Libraries	Select any third-party Python packages (.whl or .zip) here.
   Python Archives	Select any custom Python virtual environment (.zip) here.

4. Click Deploy.

For more deployment parameter details, see Deploy a job.