scala - 将带有参数的案例类作为案例类转换为 Avro 消息以发送到 Kafka

我正在使用具有嵌套案例类和 Seq[Nested Case Classes] 的案例类
问题是当我尝试使用它抛出的 KafkaAvroSerializer 序列化它时:

Caused by: java.lang.IllegalArgumentException: Unsupported Avro type. Supported types are null, Boolean, Integer, Long, Float, Double, String, byte[] and IndexedRecord
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerDe.getSchema(AbstractKafkaAvroSerDe.java:115)
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:71)
    at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:54)
    at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:60)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:879)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:841)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:728)```

最佳答案

如果您想将 Avro 与案例类等 Scala 构造一起使用，我建议您使用 Avro4s 。这对所有 scala 功能都有 native 支持，甚至可以从您的模型创建架构，如果这是您想要的。

有一些带有自动类型类派生的陷阱。这是我学到的。

至少使用 avro4s 版本 2.0.4

一些宏生成带有编译器警告的代码，并且还会破坏疣去除剂。我们必须添加以下注释才能编译我们的代码(有时错误是找不到隐式的，但它是由宏生成的代码中的错误引起的):

@com.github.ghik.silencer.silent
@SuppressWarnings(Array("org.wartremover.warts.Null", "org.wartremover.warts.AsInstanceOf", "org.wartremover.warts.StringPlusAny"))

下一个自动类型类派生一次只能工作一个级别。我创建了一个对象来保存我的架构的所有 SchemaFor 、 Decoder 和 Encoder 实例。然后我从最内部的类型开始显式地构建类型类实例。我还使用 implicitly 来验证每个 ADT 在移动到下一个之前会解决。例如:

sealed trait Notification
object Notification {
  final case class Outstanding(attempts: Int) extends Notification
  final case class Complete(attemts: Int, completedAt: Instant) extends Notification
}

sealed trait Job
final case class EnqueuedJob(id: String, enqueuedAt: Instant) extends Job
final case class RunningJob(id: String, enqueuedAt: Instant, startedAt: Instant) extends Job
final case class FinishedJob(id: String, enqueuedAt: Instant, startedAt: Instant, completedAt: Instant) extends Job

object Schema {

  // Explicitly define schema for ADT instances
  implicit val schemaForNotificationComplete: SchemaFor[Notification.Complete] = SchemaFor.applyMacro
  implicit val schemaForNotificationOutstanding: SchemaFor[Notification.Outstanding] = SchemaFor.applyMacro

  // Verify Notification ADT is defined
  implicitly[SchemaFor[Notification]]
  implicitly[Decoder[Notification]]
  implicitly[Encoder[Notification]]

  // Explicitly define schema, decoder and encoder for ADT instances
  implicit val schemaForEnqueuedJob: SchemaFor[EnqueuedJob] = SchemaFor.applyMacro
  implicit val decodeEnqueuedJob: Decoder[EnqueuedJob] = Decoder.applyMacro
  implicit val encodeEnqueuedJob: Encoder[EnqueuedJob] = Encoder.applyMacro

  implicit val schemaForRunningJob: SchemaFor[RunningJob] = SchemaFor.applyMacro
  implicit val decodeRunningJob: Decoder[RunningJob] = Decoder.applyMacro
  implicit val encodeRunningJob: Encoder[RunningJob] = Encoder.applyMacro

  implicit val schemaForFinishedJob: SchemaFor[FinishedJob] = SchemaFor.applyMacro
  implicit val decodeFinishedJob: Decoder[FinishedJob] = Decoder.applyMacro
  implicit val encodeFinishedJob: Encoder[FinishedJob] = Encoder.applyMacro

  // Verify Notification ADT is defined
  implicitly[Encoder[Job]]
  implicitly[Decoder[Job]]
  implicitly[SchemaFor[Job]]

  // And so on until complete nested ADT is defined
}

关于scala - 将带有参数的案例类作为案例类转换为 Avro 消息以发送到 Kafka，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/56172798/