我使用flink-1.0-SNAPSHOT来使用来自 kafka 的数据。数据以Snappy 压缩字节 [] 的形式传入,传递给 thrift 供以后使用。
当我使用flink检索数据时,它会以某种方式损坏或处理不当,以至于无法解压缩。代码源自此示例,如下所示:
DataStream<String> messageStream = env.addSource(new FlinkKafkaConsumer081<>(parameterTool.getRequired("topic"), new SimpleStringSchema(), parameterTool.getProperties()));
messageStream.rebalance().map(new MapFunction<String, String>() {
@Override public String map(String value) throws Exception {
boolean bvalid = Snappy.isValidCompressedBuffer(value.getBytes());
});
isValidCompressedBuffer每次都返回 false 。
当通过其他途径使用时,数据被认为是好的。
我错过了什么?
解决方案:
我发布这个是因为我找不到任何使用RawSchema
.
public static void main(String[] args) throws Exception {
// create execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// parse user parameters
ParameterTool parameterTool = ParameterTool.fromArgs(args);
DataStream<byte[]> dataStream = env.addSource(new FlinkKafkaConsumer081<>(parameterTool.getRequired("topic"), new RawSchema(), parameterTool.getProperties()));
dataStream.map(new MapFunction<byte[], Object>() {
@Override
public Object map(byte[] bytes) throws Exception {
boolean bvali = Snappy.isValidCompressedBuffer(bytes);
});
return 0;
}
}).print();
env.execute();
}