用Flink反序列化Protobuf kafka消息

Question

我试图阅读和打印从Kafka使用Apache Flink的原型消息。

我遵循官方文件，但没有成功：https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers/

Flink消费者代码是：

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.enableCheckpointing(5000)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setCheckpointStorage(s"$targetPath/checkpoints")
    env.getConfig.registerTypeWithKryoSerializer(classOf[User], classOf[ProtobufSerializer])

    val source = KafkaSource.builder[User]
      .setBootstrapServers(brokers)
      .setTopics(topic)
      .setGroupId(consumerGroupId)
      .setValueOnlyDeserializer(new ProtoDeserializer())
      .setStartingOffsets(OffsetsInitializer.earliest)
      .build

    val stream = env.fromSource(source, WatermarkStrategy.noWatermarks[User], kafkaTableName)
    stream.print()
    env.execute()
  }

反序列化器代码是：

class ProtoDeserializer extends DeserializationSchema[User] {

  override def getProducedType: TypeInformation[User] = null

  override def deserialize(message: Array[Byte]): User = User.parseFrom(message)

  override def isEndOfStream(nextElement: User): Boolean = false
}

在执行拖缆时，我遇到以下错误：

Protocol message contained an invalid tag (zero).

值得一提的是，我使用confluent protobuf consumer成功地读取和反序列化了消息，因此消息似乎没有损坏。

Answer 1

汇合protobuf序列化程序不会生成可由其他反序列化程序直接反序列化的内容。confluent的文档中描述了该格式：它以一个魔术字节（始终为零）开始，后跟一个四字节的模式ID。protobuf负载随后从字节5开始。

getProducedType方法应该返回适当的TypeInformation，在本例中是TypeInformation。of（User.class）。如果不这样做，您可能会在运行时遇到问题。

与KafkaSource一起使用的反序列化程序不需要实现isEndOfStream，但它不会造成任何影响。