Spark Streaming Kafka: ClassNotFoundException для ByteArrayDeserializer при запуске с искрой-отправкой - PullRequest
Новая
       25

Spark Streaming Kafka: ClassNotFoundException для ByteArrayDeserializer при запуске с искрой-отправкой

0 голосов
/ 04 октября 2018

Я новичок в Scala / Spark Streaming и в StackOverflow, поэтому прошу прощения за мое форматирование.Я сделал приложение Scala, которое читает файлы журнала из потока Kafka.Он отлично работает в IDE, но я буду проклят, если смогу запустить его, используя spark-submit.Всегда происходит сбой: 

ClassNotFoundException: org.apache.kafka.common.serialization.ByteArrayDeserializer

Строка, на которую ссылается Исключение, является командой загрузки в этом фрагменте: 

val records = spark
  .readStream
  .format("kafka") // <-- use KafkaSource
  .option("subscribe", kafkaTopic)
  .option("kafka.bootstrap.servers", kafkaBroker) // 192.168.4.86:9092
  .load()
  .selectExpr("CAST(value AS STRING) AS temp")
  .withColumn("record", deSerUDF($"temp"))
  • IDE: IntelliJ
  • Spark: 2.2.1
  • Scala: 2.11.8
  • Кафка: kafka_2.11-0.10.0.0

Соответствующие части pom.xml: 

<properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <encoding>UTF-8</encoding>
    <scala.version>2.11.8</scala.version>
    <scala.compat.version>2.11</scala.compat.version>
    <spark.version>2.2.1</spark.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>${scala.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>${spark.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>com.github.scala-incubator.io</groupId>
        <artifactId>scala-io-file_2.11</artifactId>
        <version>0.4.3-1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.0.0</version>
        <!-- version>2.0.0</version -->
    </dependency>

Примечание: я не думаю, что это связано, но я должен использовать zip -d BroLogSpark.jar "META-INF/*.SF" и zip -d BroLogSpark.jar "META-INF/*.DSA", чтобы понять смысл манифестных подписей.

Мой файл jar не содержит ни одного из org.apache.kafka.Я видел несколько постов, которые настоятельно рекомендуют, чтобы у меня было несоответствие в версиях, и я пробовал бесчисленное множество изменений в pom.xml и spark-submit.После каждого изменения я подтверждаю, что оно все еще выполняется в среде IDE, затем продолжаю попытки использовать spark-submit в той же системе, для того же пользователя.Ниже приведена моя последняя попытка, где мой BroLogSpark.jar находится в текущем каталоге, а «192.168.4.86:9092 профиль» являются входными аргументами. 

spark-submit --packages org.apache.spark:spark-streaming-kafka-0-10_2.11:2.2.1,org.apache.kafka:kafka-clients:0.10.0.0 BroLogSpark.jar 192.168.4.86:9092 BroFile

1 Ответ

0 голосов
/ 04 октября 2018

Добавить ниже зависимости тоже 

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.0.0</version>
</dependency>
...