kafka和spark之间的连接：找不到数据源：kafka

Question

我试图通过从一个主题读取数据并尝试将此主题的内容打印到 DataFrame 中来建立 kafka 和 spark 之间的链接，但是通过在 kafka 和 spark 之间建立连接，我得到了这个错误：

错误：

Exception in thread "main" org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".;
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:652)
    at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:161)
    at readFromKafka$.main(readFromKafka.scala:17)
    at readFromKafka.main(readFromKafka.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:845)
    at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)
    at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)
    at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
    at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:920)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:929)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我的代码内容：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, from_json}
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}

object readFromKafka
{
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder().master("spark://192.168.1.51:7077").appName("kafka_spark").getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092")
      .option("subscribe", "test")
      .option("startingOffsets", "earliest") // From starting
      .load()

    df.printSchema() }}

build.sbt：

name := "readFromSpark"

version := "1.0"

scalaVersion := "2.12.8"

val sparkVersion = "2.4.4"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion,
  "org.apache.spark" %% "spark-sql" % sparkVersion,
  "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.4.4" % Test
)

我做了一个sbt clean package，它得到以下信息，没有任何错误：

[info] welcome to sbt 1.4.7 (Private Build Java 1.8.0_282)
[info] loading project definition from /home/hamza/spark-project/spark-learning/project
[info] loading settings for project spark-learning from build.sbt ...
[info] set current project to spark-learning (in build file:/home/hamza/spark-project/spark-learning/)
[success] Total time: 0 s, completed Mar 12, 2021 1:13:50 PM
[warn] There may be incompatibilities among your library dependencies; run 'evicted' to see detailed eviction warnings.
[info] compiling 7 Scala sources to /home/hamza/spark-project/spark-learning/target/scala-2.12/classes ...
[warn] multiple main classes detected: run 'show discoveredMainClasses' to see the list
[success] Total time: 18 s, completed Mar 12, 2021 1:14:08 PM

运行火花作业：

spark-submit --master spark://192.168.1.51:7077 --class readFromKafka target/scala-2.12/spark-learning_2.12-1.0.jar

有什么帮助吗？

Answer 1

如果你想通过 spark-submit 运行代码，你需要从 Kafka 依赖中移除 % Test

您还需要设置 sbt-assembly 以确保 Kafka 依赖项实际上包含在 jar 中