无法从Kafka主题读取数据

Question

    val kafka_bootstrap_servers = "localhost:9092"

    val users_df = spark.read
      .format("kafka")
      .option("kafka.bootstrap.servers", kafka_bootstrap_servers)
      .option("subscribe", kafka_topic_name)
      .load()

    val users_df_1 = users_df.selectExpr("CAST(value AS STRING)", "CAST(timestamp AS TIMESTAMP)")

    val user_schema = StructType(
      List(
        StructField("RecordNumber", IntegerType, true),
        StructField("Zipcode", StringType, true),
        StructField("ZipCodeType", StringType, true),
        StructField("City", StringType, true),
        StructField("State", StringType, true),
        StructField("LocationType", StringType, true),
        StructField("Lat", StringType, true),
        StructField("Long", StringType, true),
        StructField("Xaxis", StringType, true),
        StructField("Yaxis", StringType, true),
        StructField("Zaxis", StringType, true),
        StructField("WorldRegion", StringType, true),
        StructField("Country", StringType, true),
        StructField("LocationText", StringType, true),
        StructField("Location", StringType, true),
        StructField("Decommisioned", StringType, true)
      )
    )

    val users_df_2 = users_df_1.select(from_json(col("RecordNumber"), user_schema)
      .as("user_detail"), col("Zipcode"))

  val users_df_3 = users_df_2.select(col = "user_detail.*", "Zipcode")

    users_df_3.printSchema()
    users_df_3.show(numRows = 10, truncate = false)

    spark.stop()
    println("Apache spark application completed.")



  }


} 

下面的json数据示例

{"RecordNumber":76511,"Zipcode":27007,"ZipCodeType":"STANDARD","City":"ASH HILL","State":"NC","LocationType":"NOT ACCEPTABLE","Lat":36.4,"Long":-80.56,"Xaxis":0.13,"Yaxis":-0.79,"Zaxis":0.59,"WorldRegion":"NA","Country":"US","LocationText":"Ash Hill, NC","Location":"NA-US-NC-ASH HILL","Decommisioned":false,"TaxReturnsFiled":842,"EstimatedPopulation":1666,"TotalWages":28876493}

下面的错误消息

线程“main”org.apache.spark.sql.analysisException中出现异常：未能找到数据源：Kafka。请按照“结构化流+Kafka集成指南”的部署部分部署应用程序。；在org.apache.spark.sql.execution.datasources.datasource$.lookupdatasource(datasource.scala:652)，在org.apache.spark.sql.dataframereader.load（dataframereader.scala:194)，在org.apache.spark.sql.dataframereader.load（dataframereader.scala:167)，在streamingapp$.main（streamingapp.scala）

需要帮助阅读Kafka主题的数据。

Answer 1

请遵循spark streaming+kafka集成指南

https://spark.apache.org/docs/latest/structure-streaming-kafka-integration.html

您可能丢失了工件“spark-sql-kafka-0-102.12”