flink 读取kafka之自定义DeserializationSchema序列化

 


import com.alibaba.fastjson.JSON;
import com.hundsun.quant.flow.model.Snapshot;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.Typeinformation;

/**
 * @author  fl
 * @date: 2022-01-05  21:30
 * @description: flink 读取kafka数据， kafka把内容是 转化成 snapshot 对象
 * <p>
 * 自定义DeserializationSchema需要实现DeserializationSchema接口，这一部分代码可以参考官方代码
 * org.apache.flink.streaming.examples.statemachine.kafka.EventDeSerializer。
 * 我需要实现的是将从kafka获取到的json数据转化为我需要的自定义pojo类（VideoData）。
 * 主要是要实现DeserializationSchema方法的deserialize方法，这个方法的输入是byte[] message类型，我们需要将其转换为String类型，
 * 然后通过JSON工具类解析成POJO类。这里我使用的是fastjson
 */
public class SnapshotSerializer implements DeserializationSchema<Snapshot> {
    @Override
    public Snapshot deserialize(byte[] message) {
        // json 转成对象
        return JSON.parSEObject(new String(message), Snapshot.class);
    }

    @Override
    public boolean isEndOfStream(Snapshot nextElement) {
        return false;
    }

    @Override
    public Typeinformation<Snapshot> getProducedType() {
        return null;
    }
}