无法流式传输来自kafka debezium连接器的avro格式化数据

mnemlml8  于 2021-06-07  发布在  Kafka
关注(0)|答案(0)|浏览(273)

我流蒙哥oplog数据通过Kafka。我使用debezium cdcKafka连接器到尾部mongo oplog。
模式注册表使用avroconverter转换器序列化键和值
bootstrap.servers服务器=localhost:9092
kafka key.converter=io.confluent.connect.avro.avroconverter key.converter.schema.registry.url=http://localhost：8081 value.converter=io.confluent.connect.avro.avroconverter value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.jsonconverter internal.value.converter=org.apache.kafka.connect.json.jsonconverter internal.key.converter.schemas.enable=false internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
下面的代码流的Kafka数据和反序列化它使用Kafka夫罗德序列化

  1. import io.confluent.kafka.schemaregistry.client.rest.RestService
  2. import io.confluent.kafka.serializers.KafkaAvroDeserializer
  3. import org.apache.avro.Schema
  4. import org.apache.spark.sql.SparkSession
  5. import scala.collection.JavaConverters._
  6. object KafkaStream{
  8.   case class DeserializedFromKafkaRecord(key: String, value: String)
  10.   def main(args: Array[String]): Unit = {
  12.     val sparkSession = SparkSession
  13.       .builder
  14.       .master("local[*]")
  15.       .appName("kafka")
  16.       .getOrCreate()
  17.     //sparkSession.sparkContext.setLogLevel("ERROR")
  19.     import sparkSession.implicits._
  21.     val schemaRegistryURL = "http://127.0.0.1:8081"
  23.     val topicName = "prodCollection.inventory.Prod"
  24.     val subjectValueName = topicName + "-value"
  26.     //create RestService object
  27.     val restService = new RestService(schemaRegistryURL)
  29.     //.getLatestVersion returns io.confluent.kafka.schemaregistry.client.rest.entities.Schema object.
  30.     val valueRestResponseSchema = restService.getLatestVersion(subjectValueName)
  32.     //Use Avro parsing classes to get Avro Schema
  33.     val parser = new Schema.Parser
  34.     val topicValueAvroSchema: Schema = parser.parse(valueRestResponseSchema.getSchema)
  36.     //key schema is typically just string but you can do the same process for the key as the value
  37.     val keySchemaString = "\"string\""
  38.     val keySchema = parser.parse(keySchemaString)
  40.     //Create a map with the Schema registry url.
  41.     //This is the only Required configuration for Confluent's KafkaAvroDeserializer.
  42.     val props = Map("schema.registry.url" -> schemaRegistryURL)
  44.     //Declare SerDe vars before using Spark structured streaming map. Avoids non serializable class exception.
  45.     var keyDeserializer: KafkaAvroDeserializer = null
  46.     var valueDeserializer: KafkaAvroDeserializer = null
  48.     //Create structured streaming DF to read from the topic.
  49.     val rawTopicMessageDF = sparkSession.readStream
  50.       .format("kafka")
  51.       .option("kafka.bootstrap.servers", "localhost:9092")
  52.       .option("subscribe", topicName)
  53.       .option("startingOffsets", "earliest")
  54.       .option("key.deserializer","KafkaAvroDeserializer")
  55.       .option("value.deserializer","KafkaAvroDeserializer")
  56.       //.option("maxOffsetsPerTrigger", 20)  //remove for prod
  57.       .load()
  58.     rawTopicMessageDF.printSchema()
  60.     //instantiate the SerDe classes if not already, then deserialize!
  61.     val deserializedTopicMessageDS = rawTopicMessageDF.map{
  62.       row =>
  63.         if (keyDeserializer == null) {
  64.           keyDeserializer = new KafkaAvroDeserializer
  65.           keyDeserializer.configure(props.asJava, true)  //isKey = true
  66.         }
  67.         if (valueDeserializer == null) {
  68.           valueDeserializer = new KafkaAvroDeserializer
  69.           valueDeserializer.configure(props.asJava, false) //isKey = false
  70.         }
  72.         //Pass the Avro schema.
  73.         val deserializedKeyString = keyDeserializer.deserialize(topicName, row.getAs[Array[Byte]]("key"), keySchema).toString //topic name is actually unused in the source code, just required by the signature. Weird right?
  74.       val deserializedValueJsonString = valueDeserializer.deserialize(topicName, row.getAs[Array[Byte]]("value"), topicValueAvroSchema).toString
  76.         DeserializedFromKafkaRecord(deserializedKeyString, deserializedValueJsonString)
  77.     }
  78.     deserializedTopicMessageDS.printSchema()
  80.       deserializedTopicMessageDS.writeStream
  81.       .outputMode("append")
  82.       .format("console")
  83.       .option("truncate", false)
  84.       .start()

反序列化的TopicMessageDS数据集架构正在根据需要进行转换，但流将被以下信息停止，

  1. root
  2.  |-- key: binary (nullable = true)
  3.  |-- value: binary (nullable = true)
  4.  |-- topic: string (nullable = true)
  5.  |-- partition: integer (nullable = true)
  6.  |-- offset: long (nullable = true)
  7.  |-- timestamp: timestamp (nullable = true)
  8.  |-- timestampType: integer (nullable = true)
  10. root
  11.  |-- key: string (nullable = true)
  12.  |-- value: string (nullable = true)
  14. 18/08/13 22:53:54 INFO StateStoreCoordinatorRef: Registered StateStoreCoordinator endpoint
  15. 18/08/13 22:53:54 INFO StreamExecution: Starting [id = b1fb3ce2-08d0-4d87-b031-af129432d91a, runId = 38b66e4a-040f-42c8-abbe-bc27fa3b9462]. Use /private/var/folders/zf/6dh44_fx1sn2dp2w7d_54wg80000gn/T/temporary-ae7a93f6-0307-4f39-ba44-93d5d3d7c0ab to store the query checkpoint.
  16. 18/08/13 22:53:54 INFO SparkContext: Invoking stop() from shutdown hook
  17. 18/08/13 22:53:54 INFO SparkUI: Stopped Spark web UI at http://192.168.0.100:4040
  18. 18/08/13 22:53:54 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
  19. 18/08/13 22:53:54 INFO MemoryStore: MemoryStore cleared
  20. 18/08/13 22:53:54 INFO BlockManager: BlockManager stopped
  21. 18/08/13 22:53:54 INFO BlockManagerMaster: BlockManagerMaster stopped
  22. 18/08/13 22:53:54 INFO ConsumerConfig: ConsumerConfig values: 
  23.     auto.commit.interval.ms = 5000
  24.     auto.offset.reset = earliest
  25.     bootstrap.servers = [localhost:9092]
  26.     check.crcs = true
  27.     client.id = 
  28.     connections.max.idle.ms = 540000
  29.     default.api.timeout.ms = 60000
  30.     enable.auto.commit = false
  31.     exclude.internal.topics = true
  32.     fetch.max.bytes = 52428800
  33.     fetch.max.wait.ms = 500
  34.     fetch.min.bytes = 1
  35.     group.id = spark-kafka-source-b9f0f64b-952d-4733-ba3e-aa753954b2ef--1115279952-driver-0
  36.     heartbeat.interval.ms = 3000
  37.     interceptor.classes = []
  38.     internal.leave.group.on.close = true
  39.     isolation.level = read_uncommitted
  40.     key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
  41.     max.partition.fetch.bytes = 1048576
  42.     max.poll.interval.ms = 300000
  43.     max.poll.records = 1
  44.     metadata.max.age.ms = 300000
  45.     metric.reporters = []
  46.     metrics.num.samples = 2
  47.     metrics.recording.level = INFO
  48.     metrics.sample.window.ms = 30000
  49.     partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
  50.     receive.buffer.bytes = 65536
  51.     reconnect.backoff.max.ms = 1000
  52.     reconnect.backoff.ms = 50
  53.     request.timeout.ms = 30000
  54.     retry.backoff.ms = 100
  55.     sasl.client.callback.handler.class = null
  56.     sasl.jaas.config = null
  57.     sasl.kerberos.kinit.cmd = /usr/bin/kinit
  58.     sasl.kerberos.min.time.before.relogin = 60000
  59.     sasl.kerberos.service.name = null
  60.     sasl.kerberos.ticket.renew.jitter = 0.05
  61.     sasl.kerberos.ticket.renew.window.factor = 0.8
  62.     sasl.login.callback.handler.class = null
  63.     sasl.login.class = null
  64.     sasl.login.refresh.buffer.seconds = 300
  65.     sasl.login.refresh.min.period.seconds = 60
  66.     sasl.login.refresh.window.factor = 0.8
  67.     sasl.login.refresh.window.jitter = 0.05
  68.     sasl.mechanism = GSSAPI
  69.     security.protocol = PLAINTEXT
  70.     send.buffer.bytes = 131072
  71.     session.timeout.ms = 10000
  72.     ssl.cipher.suites = null
  73.     ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
  74.     ssl.endpoint.identification.algorithm = https
  75.     ssl.key.password = null
  76.     ssl.keymanager.algorithm = SunX509
  77.     ssl.keystore.location = null
  78.     ssl.keystore.password = null
  79.     ssl.keystore.type = JKS
  80.     ssl.protocol = TLS
  81.     ssl.provider = null
  82.     ssl.secure.random.implementation = null
  83.     ssl.trustmanager.algorithm = PKIX
  84.     ssl.truststore.location = null
  85.     ssl.truststore.password = null
  86.     ssl.truststore.type = JKS
  87.     value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
  89. 18/08/13 22:53:54 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
  90. 18/08/13 22:53:54 INFO SparkContext: Successfully stopped SparkContext
  91. 18/08/13 22:53:54 INFO ShutdownHookManager: Shutdown hook called
  92. 18/08/13 22:53:54 INFO ShutdownHookManager: Deleting directory /private/var/folders/zf/6dh44_fx1sn2dp2w7d_54wg80000gn/T/spark-e1c2b259-39f2-4d65-9919-74ab1ad6acae
  93. 18/08/13 22:53:54 INFO ShutdownHookManager: Deleting directory /private/var/folders/zf/6dh44_fx1sn2dp2w7d_54wg80000gn/T/temporary-ae7a93f6-0307-4f39-ba44-93d5d3d7c0ab
  95. Process finished with exit code 0
apache-kafkaapache-sparkconfluent-schema-registrydebezium

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com