Jar-файл имеет класс, но я получаю java .lang.ClassNotFoundException: org. apache .kafka.clients.consumer.ConsumerRecord - PullRequest
Новая
       90

Jar-файл имеет класс, но я получаю java .lang.ClassNotFoundException: org. apache .kafka.clients.consumer.ConsumerRecord

0 голосов
/ 07 мая 2020

Я выполняю задание потоковой передачи искр для использования из kafka с использованием прямого подхода (для kafka 0.1.0 или выше). Создал файл POM с использованием maven-assembly-plugin и проверил содержимое файла jar с помощью jar tf <jar file> | grep ConsumerRecord. Я получаю следующий результат:

org / apache / kafka / clients / consumer / ConsumerRecord.class org / apache / kafka / clients / consumer / ConsumerRecords $ ConcatenatedIterable $ 1.class org / apache / kafka / clients /consumer/ConsumerRecords$ConcatenatedIterable.class org / apache / kafka / clients / consumer / ConsumerRecords.class

Но когда я запускаю задание spark-submit в моем кластере (с мастером как локальным, так и пряжи) , Я получаю следующее исключение.

java .lang.ClassNotFoundException: org. apache .kafka.clients.consumer.ConsumerRecord

Другой вариант, который я пробовал - построил закрашенную банку, используя maven-shade-plugin. Тот же результат. *) 

package com.myCompany;

import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.TaskContext;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.StreamingContext;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import scala.Tuple2;

public class ReadFromKafka {

    public static void main(String args[]) throws InterruptedException {

        SparkConf conf = new SparkConf();// .setAppName("Decryption-spark-streaming").setMaster("yarn");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

        Map<String, Object> kafkaParams = new HashMap<String, Object>();
        kafkaParams.put("bootstrap.servers", "server1:9093");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "my_cg");
        kafkaParams.put("auto.offset.reset", "earliest");
        kafkaParams.put("enable.auto.commit", false);
        kafkaParams.put("security.protocol", "SSL");
        kafkaParams.put("ssl.truststore.location", "abc.jks");
        kafkaParams.put("ssl.truststore.password", "changeit");
        kafkaParams.put("ssl.keystore.location", "abc.jks");
        kafkaParams.put("ssl.keystore.password", "changeme");
        kafkaParams.put("ssl.key.password", "changeme");

        Collection<String> topics = Arrays.asList("myTopic");

        JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jsc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));

        stream.mapToPair(record -> new Tuple2<>(record.key(), record.value()));

        stream.foreachRDD(rdd -> {
            OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
            rdd.foreachPartition(consumerRecords -> {
                OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
                System.out.println(o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());
            });
        });

        stream.foreachRDD(rdd -> {
            OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();

            // some time later, after outputs have completed
            ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);
        });

        // Start the computation
        jsc.start();
        jsc.awaitTermination();
    }}

1 Ответ

0 голосов
/ 25 мая 2020

Добавление зависимого файла jar (spark-streaming-kafka-0-10_2.11.jar) к команде spark-submit помогло решить эту проблему 

spark-submit --master yarn --deploy-mode cluster --name spark-streaming-test\
--executor-memory 1g --num-executors 4 --driver-memory 1g --jars\
/home/spark/jars/spark-streaming-kafka-0-10_2.11.jar --class\
com.mycompany.ReadFromKafka spark-streaming-test-1-jar-with-dependencies.jar
...