试图使用avro4s定义序列化程序，但得到一个缺少的隐式错误

izkcnapc 于 2021-06-06 发布在 Kafka

关注(0)|答案(2)|浏览(305)

我正在使用flink（1.7）kafka客户端和avro4s（2.0.4），我想序列化到字节数组：

class AvroSerializationSchema[IN : SchemaFor : FromRecord: ToRecord] extends SerializationSchema[IN] {
  override def serialize(element: IN): Array[Byte] = {
    val str = AvroSchema[IN]
    val schema: Schema = new Parser().parse(str.toString)
    val out = new ByteArrayOutputStream()
    val os = AvroOutputStream.data[IN].to(out).build(schema)
    os.write(element)
    out.close()
    out.flush()
    os.flush()
    os.close()
    out.toByteArray
  }
}

然而，我不断得到这个例外：

Error:(15, 35) could not find implicit value for evidence parameter of type com.sksamuel.avro4s.Encoder[IN]
    val os = AvroOutputStream.data[IN].to(out).build(schema)

以及

Error:(15, 35) not enough arguments for method data: (implicit evidence$3: com.sksamuel.avro4s.Encoder[IN])com.sksamuel.avro4s.AvroOutputStreamBuilder[IN].
Unspecified value parameter evidence$3.
    val os = AvroOutputStream.data[IN].to(out).build(schema)

apache-kafka apache-flink avro4s

来源：https://stackoverflow.com/questions/56024318/trying-to-define-serializer-using-avro4s-but-getting-a-missing-implicit-error

2条答案

按热度按时间

lmvvr0a81#

根据规范 IN 必须是 Encoder 类型：

object AvroOutputStream {

  /**
    * An [[AvroOutputStream]] that does not write the schema. Use this when
    * you want the smallest messages possible at the cost of not having the schema available
    * in the messages for downstream clients.
    */   def binary[T: Encoder] = new AvroOutputStreamBuilder[T](BinaryFormat)

  def json[T: Encoder] = new AvroOutputStreamBuilder[T](JsonFormat)

  def data[T: Encoder] = new AvroOutputStreamBuilder[T](DataFormat)
}

所以应该是这样的：

class AvroSerializationSchema[IN : Encoder] ...

赞(0）回复(0）举报 2021-06-07

zpgglvta2#

你不需要使用 FromRecord 写入输出流时。这是为那些想拥有 GenericRecord 供自己使用。你需要使用 Encoder .

class AvroSerializationSchema[IN : SchemaFor : Encoder] extends SerializationSchema[IN] {
  override def serialize(element: IN): Array[Byte] = {
    val str = AvroSchema[IN]
    val schema: Schema = new Parser().parse(str.toString)
    val out = new ByteArrayOutputStream()
    val os = AvroOutputStream.data[IN].to(out).build(schema)
    os.write(element)
    out.close()
    out.flush()
    os.flush()
    os.close()
    out.toByteArray
  }
}

赞(0）回复(0）举报 2021-06-06

我来回答

试图使用avro4s定义序列化程序，但得到一个缺少的隐式错误

2条答案

相关问题

热门标签

最新问答