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

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;


public class SnapshotSerializer implements DeserializationSchema {
    @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 getProducedType() {
        return null;
    }
}