#java #flink

GZIPOutputFormat.java

import org.apache.flink.api.java.io.TextOutputFormat;
import org.apache.flink.core.fs.Path;

import java.io.IOException;

public class GZIPOutputFormat<T> extends TextOutputFormat<T> {

    public GZIPOutputFormat(Path outputPath) {
        super(outputPath);
    }

    @Override
    public void open(int taskNumber, int numTasks) throws IOException {
        super.open(taskNumber, numTasks);
        // wrap gzip
        this.stream = new GZIPWrappedStream(this.stream, 4096, false);
    }
}

// usage
sorted.write(new GZIPOutputFormat<>(new Path(jobKey.dest_path)), jobKey.dest_path, FileSystem.WriteMode.OVERWRITE);