无法从 kafka 主题读取数据答案

【问题标题】：Unable to read data from kafka topic无法从 kafka 主题读取数据
【发布时间】：2021-09-12 08:52:59
【问题描述】：

我是卡夫卡的初学者。尝试编写 Spark 应用程序以从创建的 kafka 主题中读取数据。 Kafka topic1 已启动并运行。

下面提供的代码有问题吗：

    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。请按照《Structured Streaming + Kafka Integration Guide》的部署部分部署应用；在 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:25) 在streamingApp.main(streamingApp.scala)

需要帮助才能从 kafka 主题中读取数据。

【问题讨论】：

问题是你的设置
如果您能帮助我解决设置问题，我将不胜感激
@RanLupovich，刚刚在 src 文件夹下创建了一个 pom.xml 文件。下面提供的配置：<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql-kafka-0-10_2.11</artifactId> <version>2.2.0</version> </dependency>

标签： apache-spark apache-kafka apache-spark-sql kafka-consumer-api spark-structured-streaming

【解决方案1】：

请按照 Spark Streaming + kafka 集成指南进行操作

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

您可能缺少工件“spark-sql-kafka-0-10_2.12”

【讨论】：