如何无需打包即可在远程集群运行PyFlink程序

背景说明 #

Java Flink提供了createRemoteEnvironmentAPI，可直接将作业提交至远程JobManager执行，无需提前构建zip或jar包：

public static void main(String[] args) throws Exception {
    ExecutionEnvironment env = ExecutionEnvironment
        .createRemoteEnvironment("strato-master", "7661", "/home/user/udfs.jar");

    DataSet<String> data = env.readTextFile("hdfs://path/to/file");

    data
        .filter(new FilterFunction<String>() {
            public boolean filter(String value) {
                return value.startsWith("http://");
            }
        })
        .writeAsText("hdfs://path/to/result");

    env.execute();
}

但PyFlink的官方文档仅提及预打包提交方式，虽在StreamExecutionEnvironment的注释中提到支持RemoteStreamEnvironment，却未提供明确的创建API，且AI生成的配置代码因缺少关键逻辑无法生效。

核心结论与可行方案 #

PyFlink目前（截至Flink 1.18+）并未提供和Java完全一致的「直接序列化代码提交」能力——这是因为PyFlink的Python逻辑需要分发到TaskManager的Python环境中执行，无法像Java代码那样直接在JVM中序列化传输。但可以通过以下两种方式实现无需手动打包的远程提交：

方案1：使用pyflink run命令直接提交

本地编写好PyFlink作业代码（如my_job.py）后，直接通过命令指定远程JobManager地址，Flink会自动完成代码的上传与分发：

pyflink run --jobmanager remote://<jobmanager-host>:<jobmanager-port> --python my_job.py

这种方式无需手动打包zip/jar，Flink会自动处理当前目录下Python文件的集群分发。

方案2：代码内配置远程环境并指定代码路径

通过Configuration配置远程JobManager参数，同时指定作业代码文件路径，让Flink自动识别并分发代码：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common import Configuration

def main():
    config = Configuration()
    # 配置远程JobManager地址与端口
    config.set_string("jobmanager.rpc.address", "<jobmanager-host>")
    config.set_integer("jobmanager.rpc.port", 6123)
    # 指定当前作业的Python入口文件，Flink会自动上传该文件至集群
    config.set_string("python.files", "my_job.py")

    # 创建绑定远程配置的执行环境
    env = StreamExecutionEnvironment.get_execution_environment(config)
    
    # 编写作业逻辑
    data = env.from_collection(["http://example.com", "test", "http://flink.apache.org"])
    data.filter(lambda x: x.startswith("http://")).print()

    env.execute("Remote PyFlink Job")

if __name__ == "__main__":
    main()

为什么之前的AI代码无效？ #

之前的代码仅配置了JobManager的连接参数，但未指定Python代码的分发路径，Flink无法获取到需要执行的Python逻辑，因此无法在远程集群启动作业。

注意事项 #

• 确保本地PyFlink版本与远程集群版本完全一致，避免兼容性问题
• 若作业依赖第三方Python库，需通过--py-files参数（命令行）或python.files配置项（代码内）指定依赖包路径，或在集群TaskManager节点预先安装依赖
• Batch模式可替换为BatchExecutionEnvironment，配置逻辑与上述一致

内容的提问来源于stack exchange，提问作者Calicoder