在AWS EMR集群上使用PyFlink创建FlinkKafkaConsumer时出错

可能的原因及对应解决方案 #

1. Flink与Kafka Connector版本不匹配 #

EMR集群自带的Flink版本必须和你使用的flink-sql-connector-kafka.jar版本完全一致，否则会触发类加载错误。

• 解决步骤：
  1. 执行flink --version查看EMR上的Flink版本。
  2. 下载对应版本的Kafka Connector包（需同时匹配Flink版本和MSK的Kafka版本），比如Flink 1.15对应flink-sql-connector-kafka-1.15.4.jar。
  3. 将Jar包上传至EMR集群的/home/hadoop/目录，确保路径准确。

2. Jar包路径或权限问题 #

确认指定的Jar包路径存在且hadoop用户拥有读取权限：

• 执行ls -l /home/hadoop/flink-sql-connector-kafka.jar检查文件状态与权限。
• 若文件不存在，重新上传；若权限不足，执行chmod 644 /home/hadoop/flink-sql-connector-kafka.jar赋予读取权限。

3. MSK网络连接与安全配置问题 #

EMR集群需能正常访问MSK的Bootstrap Servers：

• 安全组配置：MSK集群的安全组要允许EMR集群所在安全组的9096端口（SSL）入站访问。
• 网络环境：确保EMR与MSK在同一个VPC内，或已配置VPC对等连接；若使用私有DNS，确认EMR集群能解析MSK的域名。
• SSL参数配置：MSK的9096端口为SSL端口，必须在properties中添加'security.protocol': 'SSL'，否则会连接失败。

4. 代码语法与参数问题 #

检查代码中的语法错误和参数配置：

• 修复换行空格问题：原代码中反斜杠后多余空格会导致语法错误，修正后代码如下：

  deserialization_schema = JsonRowDeserializationSchema.builder() \
      .type_info(type_info=Types.ROW([Types.INT(), Types.STRING()])).build()
  

• 确认bootstrap.servers地址无拼写错误；group.id尽量使用简洁名称避免潜在问题。

修正后的示例代码 #

from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer
from pyflink.datastream.formats.json import JsonRowDeserializationSchema

env = StreamExecutionEnvironment.get_execution_environment()
# 确保Jar包版本与EMR Flink版本一致
env.add_jars("file:///home/hadoop/flink-sql-connector-kafka-1.15.4.jar")

deserialization_schema = JsonRowDeserializationSchema.builder() \
    .type_info(type_info=Types.ROW([Types.INT(), Types.STRING()])).build()

kafka_consumer = FlinkKafkaConsumer(
    topics='test_source_topic',
    deserialization_schema=deserialization_schema,
    properties={
        'bootstrap.servers': 'xxxx:9096,xxxx:9096', 
        'group.id': 'python-mqtt-group',
        'security.protocol': 'SSL'  # 新增SSL配置
    })

ds = env.add_source(kafka_consumer)
ds.print()  # 添加打印验证数据读取状态
env.execute("MSK Consumer Job")

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