EC2容器中PyFlink作业无报错但无法提交的排查请求

一、代码语法错误（直接导致提交卡住无响应） #

你的PyFlink代码存在多处语法错误，Python解释器无法正确解析代码，导致提交命令无响应：

1. 字符串未闭合+依赖包名错误
   env.add_jars行的第二个JAR路径字符串未闭合，且包名多了一个横杠（flink--connector应为flink-connector），修正后：

env.add_jars(
    "file:///opt/flink/lib/flink-connector-kafka-3.1.0-1.18.jar",
    "file:///opt/flink/lib/flink-connector-kafka-clients-3.1.0.jar"
)

（同步调整kafka-clients版本与连接器版本一致，避免版本不兼容）

2. FlinkKafkaConsumer参数错误
   构造FlinkKafkaConsumer时存在非法参数名（topic name是无效变量名）、参数冗余，且topics值应为实际Kafka主题名称，修正后：

kafka_consumer = FlinkKafkaConsumer(
    topics="your-actual-kafka-topic-name",
    deserialization_schema=SimpleStringSchema(),
    properties=kafka_source_properties
)

单主题场景也可使用topic参数：

kafka_consumer = FlinkKafkaConsumer(
    topic="your-actual-kafka-topic-name",
    deserialization_schema=SimpleStringSchema(),
    properties=kafka_source_properties
)

3. 主函数调用缩进错误
   if __name__ == "__main__":下方的main()没有缩进，Python会将其视为全局代码，导致逻辑执行异常，修正后：

if __name__ == "__main__":
    main()

二、依赖版本不兼容（潜在卡住原因） #

你使用的flink-connector-kafka-3.1.0-1.18.jar与flink-connector-kafka-clients-3.7.0.jar版本差异过大，Flink Kafka连接器要求客户端版本与连接器版本匹配（通常连接器版本前两位与kafka客户端版本一致），建议替换为和连接器同版本的kafka-clients包，比如flink-connector-kafka-clients-3.1.0.jar。

三、提交命令与环境检查 #

1. 确保Docker容器内的Flink集群已正常启动，可通过./bin/flink list命令验证集群状态。
2. 提交作业时可去掉-d（后台运行）参数，查看实时输出的错误信息，方便排查：

./bin/flink run --python Jobs/test.py

3. 检查Jobs/test.py文件路径在容器内是否存在，权限是否正常。

修复后的完整代码示例 #

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors import FlinkKafkaConsumer

def main():
    # Create a StreamExecutionEnvironment
    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars(
        "file:///opt/flink/lib/flink-connector-kafka-3.1.0-1.18.jar",
        "file:///opt/flink/lib/flink-connector-kafka-clients-3.1.0.jar"
    )

    # Define the Kafka source properties
    kafka_source_properties = {
        'bootstrap.servers': 'your-kafka-bootstrap-servers',
        'group.id': 'flink-alert-tokenizer',
        'auto.offset.reset': 'earliest'
    }
 
    # Create a Kafka consumer
    kafka_consumer = FlinkKafkaConsumer(
        topics="your-kafka-topic",
        deserialization_schema=SimpleStringSchema(),
        properties=kafka_source_properties
    )

    # Add the Kafka consumer as the data source
    data_stream = env.add_source(kafka_consumer)

    # Print the data to the standard output
    data_stream.print()

    # Execute the Flink streaming job
    env.execute("PyFlink Kafka Consumer Example")

if __name__ == "__main__":
    main()

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