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;
}
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。