1

我使用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();
}
4

1 回答 1

2

将字节消息读取为字符串是不正确的。您应该按原样读取字节，然后解压缩：

public Object map(byte[] bytes) throws Exception {
    boolean bvalid = Snappy.isValidCompressedBuffer(bytes);
    ...
于 2015-10-28T06:13:03.137 回答