pyflink中指明deserialization_schema的方法是什么？

kafka_consumer = FlinkKafkaConsumer(
  topics='test',
  deserialization_schema=JsonRowDeserializationSchema.builder().type_info(
    type_info=Types.ROW_NAMED(["id", "username", "attribute1", "attribute2"],
                              [Types.INT(), Types.STRING(), Types.STRING(), Types.FLOAT()])).build(),
  properties={
    'bootstrap.servers': 'hadoop1:9092,hadoop2:9092,hadoop3:9092',
    'auto.offset.reset': 'latest'})

ds = env.add_source(kafka_consumer)