Невозможно отправить искровой фрейм данных в Kafka (java.lang.ClassNotFoundException: не удалось найти источник данных: kafka.) - PullRequest
Новая
       17

Невозможно отправить искровой фрейм данных в Kafka (java.lang.ClassNotFoundException: не удалось найти источник данных: kafka.)

0 голосов
/ 06 сентября 2018

У меня проблема с передачей данных в Kafka с помощью фрейма данных Spark.

Позвольте мне объяснить мой сценарий подробно на примере. Я хочу загрузить данные для зажигания и отправить вывод искры в kafka. Я использую Gradle 3.5 и Spark 2.3.1 & Kafka 1.0.1

Вот build.gradle 

buildscript {
ext {
    springBootVersion = '1.5.15.RELEASE'
}
repositories {
    mavenCentral()
}
dependencies {
    classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
  }
 }

apply plugin: 'scala'
apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'org.springframework.boot'

group = 'com.sample'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
compile('org.springframework.boot:spring-boot-starter')
compile ('org.apache.spark:spark-core_2.11:2.3.1')
compile ('org.apache.spark:spark-sql_2.11:2.3.1')
compile ('org.apache.spark:spark-streaming-kafka-0-10_2.11:2.3.1')
compile ('org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1')

testCompile('org.springframework.boot:spring-boot-starter-test')
}

А вот и мой код: 

package com.sample
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._

object SparkConnection {

case class emp(empid:Integer, empname:String, empsal:Float)

def main(args:Array[String]) {

    val sparkConf = new SparkConf().setAppName("Spark 
    Connection").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val dataRdd = sc.textFile("/home/sample/data/sample.txt")
    val mapRdd = dataRdd.map(row => row.split(","))
    val empRdd = mapRdd.map( row => emp(row(0).toInt, row(1), row(2).toFloat))

    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    val empDF = empRdd.toDF() 

    empDF.
    select(to_json(struct(empDF.columns.map(column):_*)).alias("value"))
    .write.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("topic", "my-kafka-topic").save()

    }

  }

Пожалуйста, игнорируйте API среды весенней загрузки в build.gradle.

После сборки пакета с использованием Gradle я могу видеть все зависимые классы, упомянутые в файле .gradle.

Но когда я запускаю код с помощью spark-submit, как 

spark-submit --class com.sample.SparkConnection spark_kafka_integration.jar

Я получаю следующую ошибку 

Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: kafka. Please find packages at http://spark.apache.org/third-party-projects.html
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:635)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:241)
    at com.iniste.SparkConnection$.main(SparkConnection.scala:29)
    at com.iniste.SparkConnection.main(SparkConnection.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:894)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:198)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:228)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:137)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: kafka.DefaultSource
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$23$$anonfun$apply$15.apply(DataSource.scala:618)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$23$$anonfun$apply$15.apply(DataSource.scala:618)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$23.apply(DataSource.scala:618)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$23.apply(DataSource.scala:618)
    at scala.util.Try.orElse(Try.scala:84)
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:618)
    ... 13 more
2018-09-05 17:41:17 INFO  SparkContext:54 - Invoking stop() from shutdown hook
2018-09-05 17:41:17 INFO  AbstractConnector:318 - Stopped Spark@51684e4a{HTTP/1.1,[http/1.1]}{0.0.0.0:4040}
2018-09-05 17:41:17 INFO  MapOutputTrackerMasterEndpoint:54 - MapOutputTrackerMasterEndpoint stopped!
2018-09-05 17:41:17 INFO  MemoryStore:54 - MemoryStore cleared
2018-09-05 17:41:17 INFO  BlockManager:54 - BlockManager stopped
2018-09-05 17:41:17 INFO  BlockManagerMaster:54 - BlockManagerMaster stopped
2018-09-05 17:41:17 INFO  OutputCommitCoordinator$OutputCommitCoordinatorEndpoint:54 - OutputCommitCoordinator stopped!
2018-09-05 17:41:17 INFO  SparkContext:54 - Successfully stopped SparkContext
2018-09-05 17:41:17 INFO  ShutdownHookManager:54 - Shutdown hook called
2018-09-05 17:41:17 INFO  ShutdownHookManager:54 - Deleting directory /tmp/spark-bd4cb4ef-3883-4c26-a93f-f355b13ef306
2018-09-05 17:41:17 INFO  ShutdownHookManager:54 - Deleting directory /tmp/spark-156dfdbd-cff4-4c70-943f-35ef403a01ed

Пожалуйста, помогите мне выйти из этой ошибки. И в некоторых блогах мне предложили использовать опцию --packages с spark-submit. Но у меня есть ограничение прокси, которое требуется для загрузки указанных пакетов. Но я не могу понять, почему spark-submit не может достать банки, которые уже доступны. Пожалуйста, поправьте меня, где я делаю неправильно.

1 Ответ

0 голосов
/ 07 сентября 2018

Как и для любых приложений Spark, для запуска приложения используется spark-submit. spark-sql-kafka-0-10_2.11 и его зависимости могут быть напрямую добавлены в spark-submit с помощью --packages, как показано ниже в вашем случае 

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 com.sample.SparkConnection spark_kafka_integration.jar

Это можно найти здесь

Однако, согласно советам cricket_007, я добавил shadowjar в ваш build.gradle Так что новый может выглядеть примерно так. 

buildscript {
    ext {
        springBootVersion = '1.5.15.RELEASE'
    }
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
    }
}

plugins {
    id "com.github.johnrengelman.shadow" version "2.0.4"
}
apply plugin: 'scala'
apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'org.springframework.boot'
apply plugin: "com.github.johnrengelman.shadow"

group = 'com.sample'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    compile('org.springframework.boot:spring-boot-starter')
    compile ('org.apache.spark:spark-core_2.11:2.3.1')
    compile ('org.apache.spark:spark-sql_2.11:2.3.1')
    compile ('org.apache.spark:spark-streaming-kafka-0-10_2.11:2.3.1')
    compile ('org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1')
    compile 'org.scala-lang:scala-library:2.11.12'
    // https://mvnrepository.com/artifact/org.apache.kafka/kafka
    //compile group: 'org.apache.kafka', name: 'kafka_2.10', version: '0.8.0'

    testCompile('org.springframework.boot:spring-boot-starter-test')
}
shadowJar{
    baseName = "spark_kafka_integration"
    zip64 true
    classifier = null
    version = null
}

Таким образом, для создания вашего jar команда будет просто: shadowJar из вашего gradle.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...