使用PyFlink Kafka连接器消费Kafka数据失败求助

1. 修复addSource方法调用错误 #

你遇到的第二个核心错误是因为**新的Unified Source API（如KafkaSource）**不兼容旧的add_source()方法，必须改用from_source()，同时需要传入Watermark策略。修改你的get_data方法：

from pyflink.datastream import WatermarkStrategy  # 新增导入

def get_data(self):
    source = KafkaSource.builder() \
        .set_bootstrap_servers("localhost:9092") \
        .set_topics("test-topic") \
        .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
        .set_value_only_deserializer(SimpleStringSchema()) \
        .build()
    # 替换add_source为from_source，补充Watermark策略
    self.env \
        .from_source(source, WatermarkStrategy.no_watermarks(), "Kafka-Source") \
        .print()
    self.env.execute("source")

2. 确保Kafka依赖正确加载 #

第一个错误是因为依赖Jar未被Flink上下文正确识别，推荐两种可靠加载方式：

方式一：命令行传参加载（最稳妥）

执行Python脚本时，直接通过--jarfile指定Jar的绝对路径：

python your_script.py --jarfile file:///absolute/path/to/flink-sql-connector-kafka-1.17.1.jar

避免使用相对路径，防止Flink解析路径失败

方式二：代码中配置绝对路径

如果要在代码中指定Jar路径，确保pipeline.jars或add_jars传入绝对路径：

# 方式1：通过Configuration设置
self.config.set_string("pipeline.jars", "file:///Users/xxx/jars/flink-sql-connector-kafka-1.17.1.jar")

# 方式2：用add_jars
self.env.add_jars("file:///absolute/path/to/flink-sql-connector-kafka-1.17.1.jar")

不要将Jar包放到虚拟环境的pyflink/lib目录，这种方式无法被Flink执行上下文正确识别

额外注意事项 #

• 确保flink-sql-connector-kafka-1.17.1.jar版本与你的Flink版本（1.17.1）完全一致，版本不匹配会导致类加载异常。
• 本地测试时确认localhost:9092可以正常访问Kafka集群，避免网络连通性问题。

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