PyFlink Datastream API运行报错：找不到FlinkKafkaConsumer Java类

环境版本 #

• Python: 3.8.10
• Apache-Flink: 1.18.0
• Apache-Kafka: 3.5.1

问题描述 #

运行PyFlink Datastream API代码时抛出错误：

Could not found the Java class 'org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer'

已在代码中导入FlinkKafkaConsumer模块，但运行时仍出现上述报错。

代码示例 #

import io
import os
import gzip
import requests
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.datastream.functions import KeyedProcessFunction
from pyflink.datastream import DataStream
from pyflink.datastream import TimeCharacteristic

env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

deserialization_schema = SimpleStringSchema()

kafka_consumer = FlinkKafkaConsumer(
    topics='test_source_topic',
    deserialization_schema=deserialization_schema,
    properties={
        'bootstrap.servers': 'localhost:9092', 
        'group.id': 'test-consumer-group', 
        'auto.offset.reset': 'earliest'
        }
    )

source = env.add_source(kafka_consumer)

完整报错栈 #

Traceback (most recent call last):
  File "/root/flink_scripts/logs_05.py", line 18, in <module>
    kafka_consumer = FlinkKafkaConsumer(
  File "/usr/local/lib/python3.8/dist-packages/pyflink/datastream/connectors/kafka.py", line 203, in __init__
    j_flink_kafka_consumer = _get_kafka_consumer(topics, properties, deserialization_schema,
  File "/usr/local/lib/python3.8/dist-packages/pyflink/datastream/connectors/kafka.py", line 161, in _get_kafka_consumer
    j_flink_kafka_consumer = j_consumer_clz(topics,
  File "/usr/local/lib/python3.8/dist-packages/pyflink/util/exceptions.py", line 185, in wrapped_call
    raise TypeError(
TypeError: Could not found the Java class 'org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'

org.apache.flink.client.program.ProgramAbortException: java.lang.RuntimeException: Python process exits with code: 1
    at org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:140)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:851)
    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
    at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1095)
    at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)
    at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
    at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)
    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)
Caused by: java.lang.RuntimeException: Python process exits with code: 1
    at org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:130)
    ... 14 more

解决方案 #

报错核心原因是PyFlink运行时缺少Kafka连接器的Java依赖包，Python侧的导入只是加载了封装层，实际执行需要对应的Java类支持。以下是两种解决方法：

方法1：运行时通过命令行指定依赖Jar包 #

获取与Flink版本匹配的Kafka连接器Jar包（Flink 1.18.0对应flink-connector-kafka-1.18.0.jar）和对应版本的kafka-clients-3.5.1.jar，运行代码时添加--jarfile参数：

flink run --jarfile /path/to/flink-connector-kafka-1.18.0.jar,/path/to/kafka-clients-3.5.1.jar /root/flink_scripts/logs_05.py

如果是本地模式，也可以直接把这两个Jar包复制到Flink安装目录的lib文件夹下，重启Flink后无需每次指定参数即可运行代码。

方法2：在代码中配置依赖Jar包 #

初始化StreamExecutionEnvironment后，添加配置指定Jar包路径：

env = StreamExecutionEnvironment.get_execution_environment()
# 添加依赖Jar包配置
env.add_jars("file:///path/to/flink-connector-kafka-1.18.0.jar", "file:///path/to/kafka-clients-3.5.1.jar")
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

注意路径需使用file://前缀，确保Flink能识别本地文件路径。

额外注意事项 #

• 必须保证Kafka连接器版本与Flink版本严格匹配，版本不兼容会引发其他异常。
• 集群模式下，需确保所有TaskManager节点的lib目录都包含这些Jar包，或通过集群配置统一加载依赖。

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