Próbuję przeczytać dane AVRO z Kafki za pomocą strumieniowego strumieniowego, ale otrzymuję następujący komunikat o błędzie:

Streaming Query Exception caught!: org.apache.spark.sql.streaming.StreamingQueryException: Job aborted.
=== Streaming Query ===
Identifier: [id = 8b54c92d-6bbc-4dbc-84d0-55b762c21ba2, runId = 4bc92b3c-343e-4886-b0bc-0777b89f9ec8]
Current Committed Offsets: {KafkaV2[Subscribe[customer-avro4]]: {"customer-avro":{"0":17}}}
Current Available Offsets: {KafkaV2[Subscribe[customer-avro4]]: {"customer-avro":{"0":20}}}

Current State: ACTIVE
Thread State: RUNNABLE

Jakiś pomysł na temat tego, co może być problem i jak go rozwiązać? Kod jest następujący (inspirowany z blogem strukturalnym-strumieniowym Xebia-France). Właściwie myślę już wcześniej, ale teraz jest problem.

import com.databricks.spark.avro.SchemaConverters
import io.confluent.kafka.schemaregistry.client.{CachedSchemaRegistryClient, SchemaRegistryClient}
import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.StreamingQueryException

object AvroConsumer {
  private val topic = "customer-avro4"
  private val kafkaUrl = "http://localhost:9092"
  private val schemaRegistryUrl = "http://localhost:8081"

  private val schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 128)
  private val kafkaAvroDeserializer = new AvroDeserializer(schemaRegistryClient)

  private val avroSchema = schemaRegistryClient.getLatestSchemaMetadata(topic + "-value").getSchema
  private val sparkSchema = SchemaConverters.toSqlType(new Schema.Parser().parse(avroSchema))

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .appName("ConfluentConsumer")
      .master("local[*]")
      .getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

    spark.udf.register("deserialize", (bytes: Array[Byte]) =>
      DeserializerWrapper.deserializer.deserialize(bytes)
    )

    val kafkaDataFrame = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafkaUrl)
      .option("subscribe", topic)
      .load()

    val valueDataFrame = kafkaDataFrame.selectExpr("""deserialize(value) AS message""")

    import org.apache.spark.sql.functions._

    val formattedDataFrame = valueDataFrame.select(
      from_json(col("message"), sparkSchema.dataType).alias("parsed_value"))
      .select("parsed_value.*")

    val writer = formattedDataFrame
      .writeStream
      .format("parquet")
      .option("checkpointLocation", "hdfs://localhost:9000/data/spark/parquet/checkpoint")
    while (true) {
      val query = writer.start("hdfs://localhost:9000/data/spark/parquet/total")

      try {
        query.awaitTermination()
      }
      catch {
        case e: StreamingQueryException => println("Streaming Query Exception caught!: " + e);
      }
    }
  }

  object DeserializerWrapper {
    val deserializer: AvroDeserializer = kafkaAvroDeserializer
  }

  class AvroDeserializer extends AbstractKafkaAvroDeserializer {
    def this(client: SchemaRegistryClient) {
      this()
      this.schemaRegistry = client
    }

    override def deserialize(bytes: Array[Byte]): String = {
      val genericRecord = super.deserialize(bytes).asInstanceOf[GenericRecord]
      genericRecord.toString
    }
  }

}
0
Corram 31 lipiec 2020, 15:35

1 odpowiedź

Najlepsza odpowiedź

Pomyślał o tym - problem nie był tak, jak myślałem bezpośrednio z integracją Spark-Kafka, ale z informacją punktu kontrolnego w systemie plików HDFS. Usuwanie i odtwarzanie folderu punktu kontrolnego w HDFS rozwiązało go dla mnie.

0
Corram 31 lipiec 2020, 15:58