Flink Python DataStream API Kafka消费者-NoclassDeffoundError bytearrayDeserializer错误

Question

我在pyflink的PY4J侧有一个错误。代码下面：

env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:/" + os.getcwd() + "/jar_files/" + "flink-sql-connector-kafka-1.15.0.jar")
type_info = Types.ROW_NAMED(['id', 't', 'l', 't', 'm', "a", "e"],
                                [Types.STRING(), Types.LIST(Types.STRING()), Types.STRING(), Types.FLOAT(),
                                 Types.STRING(), Types.FLOAT(),
                                 Types.STRING()])
json_row_schema = JsonRowDeserializationSchema.builder().type_info(type_info).build()
kafka_props = {'bootstrap.servers': 'localhost:9092'}
kafka_consumer = FlinkKafkaConsumer("test", json_row_schema, kafka_props)
kafka_consumer.set_start_from_earliest()
## Stream
type_info = Types.ROW([Types.STRING(), Types.PICKLED_BYTE_ARRAY()])
serialization_schema = JsonRowSerializationSchema.Builder() \
        .with_type_info(type_info) \
        .build()
kafka_producer = FlinkKafkaProducer(
        topic='testmodels',
        serialization_schema=serialization_schema,
        producer_config={'bootstrap.servers': 'localhost:9092'}
    )

错误是：

py4j.protocol.Py4JJavaError: An error occurred while calling None.org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.
: java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer
    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.setDeserializer(FlinkKafkaConsumer.java:323)
..more..

我也有一个虚拟映射函数，在这里与之无关，因为

def _get_kafka_consumer(topics, properties, deserialization_schema, j_consumer_clz):
    if not isinstance(topics, list):
        topics = [topics]
    gateway = get_gateway()
    j_properties = gateway.jvm.java.util.Properties()
    for key, value in properties.items():
        j_properties.setProperty(key, value)

    j_flink_kafka_consumer = j_consumer_clz(topics,
                                            deserialization_schema._j_deserialization_schema,
                                            j_properties)
    return j_flink_kafka_consumer

我尝试的错误是：

• 添加env.add_jars（“ file：/” + os.getCwd （） +“/jar_files/” +“ kafka-clients-3.2.0.jar”） to代码。
• 将另一个java_import添加为java_import（gateway.jvm，“ org.apache.kafka.common.serialization。*”）在`import_flink_view'方法中。
• 使用flink-connector-kafka-1.15.0.jar和kafka-clients-3.2.0.jar一起。

他们没有解决这个问题。

Answer 1

pyflink在〜/miniconda3/envs/end＆lt; env_name＆gt;/lib/lt; pythonversion＆gt;/site_packages/pyflink/lib/code>下使用.jar文件。移动flink-connector-kafka-1.15.0.jar和kafka-clients-3.2.0.jar在此文件夹下可能会解决您的问题。