Develop Python Jobs

Choosing between Table API/SQL and DataStream API

PyFlink supports two development approaches: Table API/SQL and DataStream API. Table API/SQL is recommended for the following reasons:

• Better performance: The optimized execution plan of Table API/SQL runs entirely within the JVM. In contrast, DataStream API requires row-by-row serialization and deserialization between the JVM and the Python process, which introduces significant performance overhead.

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

• Community recommendation: The Apache Flink community prioritizes Table API/SQL as the primary development direction for PyFlink.

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

Development references

You can refer to the following documents to develop Flink business code locally. After development is complete, upload the code to the Flink development console and deploy the job.

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

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

Project structure

The recommended project structure for Python jobs is as follows:

my-flink-python-project/
├── my_job.py                # 主作业文件
├── udfs.py                  # 自定义函数（可选）
├── requirements.txt         # 第三方Python依赖（可选）
└── config.properties        # 配置文件（可选）

Dependency management

For information about how to use custom Python virtual environments, third-party Python packages, JAR files, and data files in Python jobs, see Use Python dependencies.

User-defined functions (UDFs)

The following is a development example of a Python UDSF that masks sensitive string data:

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

@udf(result_type=DataTypes.STRING())def mask_phone(phone: str):"""手机号脱敏：保留前3位和后4位，中间用****替代"""if phone is None or len(phone) != 11:return phone
    return phone[:3] + '****' + phone[7:]

Use this 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;

For information about how to register, update, and delete UDFs, see Manage UDFs.

Use connectors

For the list of connectors supported by Flink, see Supported connectors. To use a connector, perform the following steps:

1. Log on to the Realtime Compute console.

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

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

4. Click Upload Resource and select the Python package of the target connector to upload.

   You can upload a self-developed connector or a connector provided by Flink. To download the official Python packages of connectors provided by Flink, see Connector list.

5. On the O&M Center > Job O&M page, click Create Deployment > Python Deployment. Select the Python package of the target connector for the Additional Dependencies field, configure other parameters, and deploy the job.

6. Click the name of the deployed job. In the Deployment Details tab, in the Running Parameters section, click Edit. In Additional Configuration, add the location information of the Python connector packages.

   If your job depends on multiple connector Python packages, for example, two packages named connector-1.jar and connector-2.jar, the configuration is 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 and later only), add the configuration in the Running Parameters section under Additional Configuration of the job. Example:

   ## 多个连接器使用
   pipeline.used-builtin-connectors: kafka;sls
   ## 传输数据的多种数据格式
   pipeline.used-builtin-formats: avro;parquet
   ## 使用了多个已经创建的Catalog
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

For details about how to use connectors, see Complete sample code.

Job debugging

import logging

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

Local debugging

Because Realtime Compute for Apache Flink does not have Internet access by default, your code may not be able to directly connect to online data sources for testing locally. We recommend the following approaches for local debugging:

• Unit testing: Perform independent unit tests on UDFs to ensure that the function logic is correct.

• Local execution: Use local data sources (such as files or in-memory data) to simulate inputs, and run the job locally to verify the processing logic. Example:

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()# 使用本地数据源进行测试
  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 by connecting to online data sources, see Run and debug jobs with connectors locally.

Job deployment

After the Python job development is complete, upload the job to the Realtime Compute console for deployment. Perform the following steps:

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

2. In the left-side navigation pane, click File Management and upload the Python job file (.py or .zip). If there are third-party dependencies or configuration files, upload them as well.

3. On the O&M Center > Job O&M page, click Create Deployment > Python Deployment and fill in the deployment information.

   Parameter	Description
   Python File Path	Select the uploaded Python job file.
   Entry Module	If the job file is a .py file, this field is not required. If the job file is a .zip file, specify the entry module name, for example, my_job.
   Additional Dependencies	Select connector JAR files or configuration files if applicable.
   Python Libraries	Select third-party Python packages (.whl or .zip) if applicable.
   Python Archives	Select a custom Python virtual environment (.zip) if applicable.

4. Click Deploy.

For more information about deployment parameters, see Deploy a job.