如何以byte[]
格式读取KAFKA的数据?
我有一个用SimpleStringSchema()
读为String
事件的实现,但我找不到将数据读取为byte[]
的模式。
这是我的代码:
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "kafka1:9092");
properties.setProperty("zookeeper.connect", "zookeeper1:2181");
properties.setProperty("group.id", "test");
properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
properties.setProperty("auto.offset.reset", "earliest");
DataStream<byte[]> stream = env
.addSource(new FlinkKafkaConsumer010<byte[]>("testStr", ? ,properties));
最后我发现:
DataStream<byte[]> stream = env
.addSource(new FlinkKafkaConsumer010<>("testStr", new AbstractDeserializationSchema<byte[]>() {
@Override
public byte[] deserialize(byte[] bytes) throws IOException {
return bytes;
}
}, properties));
对于Scala,您应该写如下
new AbstractDeserializationSchema[Array[Byte]](){
override def deserialize(bytes: Array[Byte]): Array[Byte] = bytes
}