Kafka:=线程“main”组织中的异常。阿帕奇。火花sql。流动。StreamingQueryException:找不到连接的条目

Question

同时运行Kafka代码

1）错误流执行：查询[id=c6426655-446f-4306-91ba-d78e68e05c15， runId=420382c1-8558-45a1-b26d-f6299044fa04]终止与错误java.lang.ExceptionIn初始azerError

2） 线程“针对[id=c6426655-446f-4306-91ba-d78e68e05c15，runId=420382c1-8558-45a1-b26d-f6299044fa04]的流执行线程”中出现异常。lang.Exception初始化错误

3） 线程“main”组织中出现异常。阿帕奇。火花sql。流动。StreamingQueryException:空

sbt依赖

// https://mvnrepository.com/artifact/org.apache.spark/spark-core libraryDependencies=“org.apache.spark”%%“spark核心”%%“2.2.3”

// https://mvnrepository.com/artifact/org.apache.spark/spark-sql libraryDependencies=“org.apache.spark”%%“spark sql”%%“2.2.3”

// https://mvnrepository.com/artifact/org.apache.spark/spark-streaming libraryDependencies=“org.apache.spark”%%“spark流媒体”%%“2.2.3”%”已提供

// https://mvnrepository.com/artifact/org.apache.kafka/kafka libraryDependencies=“org.apache.kafka”%%“kafka”%%“2.1.1”

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients libraryDependencies=“org.apache.kafka”%”Kafka客户端“%”2.1.1

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams libraryDependencies=“org.apache.kafka”%”Kafka流“%”2.1.1

// https://mvnrepository.com/artifact/org.apache.spark/spark-sql-kafka-0-10 libraryDependencies=“org.apache.spark”%%”spark-sql-kafka-0-10“%%”2.2.3

//https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams-scala库依赖="org.apache.kafka " %% "kafka流-scala"%"2.1.1"

import java.sql.Timestamp

import org.apache.spark.sql.SparkSession


object demo1 {

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

    System.setProperty("hadoop.home.dir","c:\\hadoop\\")

    val spark: SparkSession = SparkSession.builder
      .appName("My Spark Application")
      .master("local[*]")
      .config("spark.sql.warehouse.dir", "file:///C:/temp") // Necessary to work around a Windows bug in Spark 2.0.0; omit if you're not on Windows.
      .config("spark.sql.streaming.checkpointLocation", "file:///C:/checkpoint")
      .getOrCreate

    spark.sparkContext.setLogLevel("ERROR")

    spark.conf.set("spark,sqlshuffle.partations","2")

    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "162.244.80.189:9092")
      .option("startingOffsets", "earliest")
      .option("group.id","test1")
      .option("subscribe", "demo11")
      .load()

    import spark.implicits._


    val dsStruc = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp").as[(String, String, Timestamp)]


    val abc = df.writeStream
    .outputMode("append")
    .format("console")
    .start().awaitTermination()

    df.show()

Answer 1

我也有同样的问题。我使用了错误的spark sql kafka库版本（2.2.0而不是2.3.0）。我的成功配置是：

组织。阿帕奇。spark-core_2.11 2.3。提供0

组织。阿帕奇。spark-sql_2.11 2.3。0

组织。阿帕奇。spark spark-sql-kafka-0-10_2.11 2.3。0

组织。阿帕奇。KafkaKafka客户端0.10。1

我希望有帮助。这篇文章给了我灵感

https://community.hortonworks.com/content/supportkb/222428/error-microbatchexecution-query-id-567e4e77-9457-4.html