Configure FileSink to work with .gz files on S3

Viewed 46

I do upgrade from flink 1.11 to 1.15.1 writing .gz files on S3. Since connectors.fs is now obsolete, I have to change the sink I've used before: EventuallyConsistentBucketingSink:

https://github.com/sjwiesman/flink/blob/83a6400e2587b067d08a64bc7e10edd4b57e71b4/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/consistent/EventuallyConsistentBucketingSink.java

to something new, so I use the FileSink:

return FileSink.forRowFormat(
                new Path(outputBasePath), new Encoder<T>() {
                    @Override
                    public void encode(T record, OutputStream stream)
                        throws IOException {
                        GzipParameters params = new GzipParameters();
                        params.setCompressionLevel(Deflater.BEST_COMPRESSION);
                        GzipCompressorOutputStream out = new GzipCompressorOutputStream(stream, params);
                        OBJECT_MAPPER.writeValue(out, record);
                        out.finish();
                    }

                })
            .withBucketAssigner(new BasePathBucketAssigner<>())
            .withRollingPolicy(OnCheckpointRollingPolicy.build())
            .build();

where outputBasePath is some s3:// URI. I got this error:

Exception in thread "main" java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS
    at org.apache.flink.runtime.fs.hdfs.HadoopRecoverableWriter.<init>(HadoopRecoverableWriter.java:60)
    at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.createRecoverableWriter(HadoopFileSystem.java:215)
    at org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createBucketWriter(FileSink.java:475)
    at org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer(FileSink.java:466)
    at org.apache.flink.connector.file.sink.FileSink.getCommittableSerializer(FileSink.java:175)
    at org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo.toString(CommittableMessageTypeInfo.java:120)

What I'm missing here?

1 Answers

If you are using hadoop, it is advised to use "s3a://" as the schema.

My source is here

Related