Amazon S3 #

Amazon Simple Storage Service (Amazon S3) provides cloud object storage for a variety of use cases. You can use S3 with Flink for reading and writing data as well in conjunction with the streaming state backends.

You can use S3 objects like regular files by specifying paths in the following format:

s3://<your-bucket>/<endpoint>

The endpoint can either be a single file or a directory, for example:

// Read from S3 bucket
FileSource<String> fileSource = FileSource.forRecordStreamFormat(
            new TextLineInputFormat(), new Path("s3://<bucket>/<endpoint>")
    ).build();
env.fromSource(
    fileSource,
    WatermarkStrategy.noWatermarks(),
    "s3-input"
);

// Write to S3 bucket
stream.sinkTo(
        FileSink.forRowFormat(
            new Path("s3://<bucket>/<endpoint>"), new SimpleStringEncoder<>()
        ).build()
);

// Use S3 as checkpoint storage
Configuration config = new Configuration();
config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "s3://<your-bucket>/<endpoint>");
env.configure(config);

Note that these examples are not exhaustive and you can use S3 in other places as well, including your high availability setup or the EmbeddedRocksDBStateBackend; everywhere that Flink expects a FileSystem URI (unless otherwise stated).

S3 FileSystem Implementations #

Flink provides three independent S3 filesystem implementations:

Previously, users had to choose between Presto (recommended for checkpointing throughput) and Hadoop (the only implementation with RecoverableWriter, required by the FileSink). The Native S3 implementation unifies both capabilities in a single plugin and measurements show significant checkpoint throughput improvements over the Presto implementation.

All three are self-contained with no dependency footprint, so there is no need to add Hadoop to the classpath to use them.

Common Configuration #

Configure Access Credentials #

After setting up the S3 FileSystem implementation, you need to make sure that Flink is allowed to access your S3 buckets. The following three approaches are independent alternatives — choose the one that fits your environment:

Identity and Access Management (IAM) (Recommended) #

The recommended way of setting up credentials on AWS is via Identity and Access Management (IAM). You can use IAM features to securely give Flink instances the credentials that they need to access S3 buckets. Details about how to do this are beyond the scope of this documentation. Please refer to the AWS user guide. What you are looking for are IAM Roles.

If you set this up correctly, you can manage access to S3 within AWS and don’t need to distribute any access keys to Flink.

Delegation Tokens #

Delegation tokens provide time-bounded, automatically negotiated credentials. The JobManager uses long-lived credentials (access key and secret key) to call AWS STS and obtain short-lived session tokens, which are then automatically distributed to TaskManagers.

Each S3 implementation has its own delegation token provider with a dedicated configuration prefix. You must set the access-key, secret-key, and region under the corresponding prefix for the implementation you are using:

# For Native S3 implementation
security.delegation.token.provider.s3-native.access-key: your-access-key
security.delegation.token.provider.s3-native.secret-key: your-secret-key
security.delegation.token.provider.s3-native.region: us-east-1

# For Hadoop implementation
security.delegation.token.provider.s3-hadoop.access-key: your-access-key
security.delegation.token.provider.s3-hadoop.secret-key: your-secret-key
security.delegation.token.provider.s3-hadoop.region: us-east-1

# For Presto implementation
security.delegation.token.provider.s3-presto.access-key: your-access-key
security.delegation.token.provider.s3-presto.secret-key: your-secret-key
security.delegation.token.provider.s3-presto.region: us-east-1

All three values (access-key, secret-key, region) must be set for delegation tokens to be issued. The DynamicTemporaryAWSCredentialsProvider is automatically included in the credentials provider chain for each implementation, so TaskManagers will consume the distributed tokens without additional configuration.

Access Keys #

Access to S3 can be granted via your access and secret key pair. While access keys are not inherently insecure, IAM roles are preferred as they avoid the need to manage and distribute static credentials. See the introduction of IAM roles for more context.

You need to configure both s3.access-key and s3.secret-key in Flink’s configuration file:

s3.access-key: your-access-key
s3.secret-key: your-secret-key

Configure Non-S3 Endpoint #

The S3 filesystems also support using S3 compliant object stores. To do so, configure your endpoint in Flink configuration file:

s3.endpoint: your-endpoint-hostname

Configure Path Style Access #

Some S3 compliant object stores might not have virtual host style addressing enabled by default. In such cases, you will have to provide the property to enable path style access in Flink configuration file:

s3.path-style-access: true

The legacy configuration key s3.path.style.access is still supported as a fallback for backward compatibility.

Implementation Details #

Native S3 FileSystem (Experimental) #

Experimental: The Native S3 FileSystem is experimental in Flink 2.3. It is functionally complete and has demonstrated strong performance in benchmarks.

The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2 completely removing the dependency on Hadoop. It is registered under the schemes s3:// and s3a://. It provides a drop-in replacement for the Presto and Hadoop implementations, supporting checkpointing, the FileSink (via RecoverableWriter), server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role assumption, entropy injection, and bulk copy via S3TransferManager.

Setup #

To use the Native S3 FileSystem, copy the JAR file from the opt directory to the plugins directory:

mkdir -p ./plugins/s3-fs-native
cp ./opt/flink-s3-fs-native-2.3.0.jar ./plugins/s3-fs-native/

Configuration #

In addition to the common configuration options (s3.access-key, s3.secret-key, s3.endpoint, s3.path-style-access), the Native S3 FileSystem supports the following options:

# Server-side encryption
s3.sse.type: sse-s3         # or sse-kms, aws:kms, AES256, none (default)
s3.sse.kms.key-id: arn:aws:kms:region:account:key/id   # Required for SSE-KMS

# IAM role assumption for cross-account access
s3.assume-role.arn: arn:aws:iam::account:role/RoleName
s3.assume-role.external-id: external-id-if-required
s3.assume-role.session-name: flink-s3-session
s3.assume-role.session-duration: 3600

# Performance tuning
s3.upload.min.part.size: 5242880        # 5 MB default
s3.upload.max.concurrent.uploads: 4     # Based on CPU cores
s3.read.buffer.size: 262144             # 256 KB default
s3.async.enabled: true                  # Async read/write operations
s3.bulk-copy.enabled: true              # Bulk copy via S3TransferManager
s3.bulk-copy.max-concurrent: 16         # Max concurrent copy ops

When fs.s3.aws.credentials.provider is not set, the Native S3 FileSystem automatically builds a credentials chain in the following order: delegation tokens, static credentials (if s3.access-key and s3.secret-key are configured), and the AWS SDK v2 DefaultCredentialsProvider (environment variables, instance profiles, etc.). You only need to set this option if you require a custom provider chain.

Presto S3 FileSystem #

You don’t have to configure this manually if you are running Flink on EMR.

The Presto S3 FileSystem is based on code from the Presto project. It is registered under the schemes s3:// and s3p://. It is the production-proven choice for checkpointing to S3. It does not support the FileSink (createRecoverableWriter throws UnsupportedOperationException).

Setup #

To use the Presto S3 FileSystem, copy the JAR file from the opt directory to the plugins directory:

mkdir -p ./plugins/s3-fs-presto
cp ./opt/flink-s3-fs-presto-2.3.0.jar ./plugins/s3-fs-presto/

Configuration #

The common configuration options apply. In addition, Presto-specific keys are supported via the Presto file system configuration.

Hadoop S3 FileSystem #

The Hadoop S3 FileSystem is based on code from the Hadoop Project. It is registered under the schemes s3:// and s3a://. It is the only stable implementation that supports the FileSink (via RecoverableWriter).

Setup #

To use the Hadoop S3 FileSystem, copy the JAR file from the opt directory to the plugins directory:

mkdir -p ./plugins/s3-fs-hadoop
cp ./opt/flink-s3-fs-hadoop-2.3.0.jar ./plugins/s3-fs-hadoop/

Configuration #

The common configuration options apply. In addition, Hadoop’s s3a configuration keys are supported. Hadoop configuration keys are automatically translated — for example, fs.s3a.connection.maximum becomes s3.connection.maximum.

Using Multiple S3 Implementations #

All three S3 implementations register as handlers for the s3:// scheme. Additionally, each implementation supports alternative schemes:

It is safe to load multiple S3 plugin JARs simultaneously — the priority mechanism ensures only one factory handles each scheme. The Native S3 implementation has the lowest priority (-1 vs the default 0), so when another implementation is present, it will take precedence for all overlapping schemes (e.g., s3:// and s3a://). You can override factory priorities via the fs.<scheme>.priority.<factoryClassName> configuration option.

You can use multiple S3 implementations simultaneously by leveraging their different URI schemes. For example, if a job uses the FileSystem sink with Hadoop but Presto for checkpointing:

• Use s3a:// scheme for the sink (Hadoop)
• Use s3p:// scheme for checkpointing (Presto)

The Native S3 implementation does not introduce a new URI scheme. It supports the existing s3:// and s3a:// schemes. Since both the Native S3 and Hadoop implementations register for the same schemes, Flink uses a priority-based mechanism to select which factory handles each scheme. By default, Native S3 has the lowest priority and will not be selected when another implementation is present for the same scheme.

To use the Native S3 implementation, either place only the flink-s3-fs-native plugin JAR in the plugins directory, or use the fs.<scheme>.priority.<factoryClassName> configuration to raise its priority while other implementations are present in plugins.

Advanced Features #

Entropy Injection #

All S3 file systems support entropy injection, a technique to improve the scalability of AWS S3 buckets through adding random characters near the beginning of the key.

If entropy injection is activated, a configured substring in the path is replaced with random characters. For example, path s3://my-bucket/_entropy_/checkpoints/dashboard-job/ would be replaced by something like s3://my-bucket/gf36ikvg/checkpoints/dashboard-job/. This only happens when the file creation passes the option to inject entropy! Otherwise, the file path removes the entropy key substring entirely. See FileSystem.create(Path, WriteOption) for details.

The Flink runtime currently passes the option to inject entropy only to checkpoint data files. All other files, including checkpoint metadata and external URI, do not inject entropy to keep checkpoint URIs predictable.

To enable entropy injection, configure the entropy key and the entropy length parameters.

s3.entropy.key: _entropy_
s3.entropy.length: 4 (default)

The s3.entropy.key defines the string in paths that is replaced by the random characters. Paths that do not contain the entropy key are left unchanged. If a file system operation does not pass the “inject entropy” write option, the entropy key substring is simply removed. The s3.entropy.length defines the number of random alphanumeric characters used for entropy.

s5cmd #

Supported by: Presto S3 FileSystem, Hadoop S3 FileSystem

Both flink-s3-fs-hadoop and flink-s3-fs-presto can be configured to use the s5cmd tool for faster file upload and download. Benchmark results are showing that s5cmd can be over 2 times more CPU efficient. Which means either using half the CPU to upload or download the same set of files, or doing that twice as fast with the same amount of available CPU.

In order to use this feature, the s5cmd binary has to be present and accessible to the Flink’s task managers, for example via embedding it in the used docker image. Secondly, the path to the s5cmd has to be configured via:

s3.s5cmd.path: /path/to/the/s5cmd

Configuration #

The remaining configuration options (with their default value listed below) are:

# Extra arguments that will be passed directly to the s5cmd call. Please refer to the s5cmd's official documentation.
s3.s5cmd.args: -r 0
# Maximum size of files that will be uploaded via a single s5cmd call.
s3.s5cmd.batch.max-size: 1024mb
# Maximum number of files that will be uploaded via a single s5cmd call.
s3.s5cmd.batch.max-files: 100

Both s3.s5cmd.batch.max-size and s3.s5cmd.batch.max-files control resource usage of the s5cmd binary to prevent it from overloading the task manager.

It is recommended to first configure and verify that Flink works without using s5cmd, then enable this feature.

Credentials #

If you are using access keys, they will be passed to s5cmd. Apart from that, s5cmd has its own independent way of using credentials.

Limitations #

Currently, flink-s3-fs-hadoop and flink-s3-fs-presto use s5cmd only during recovery, when downloading state files from S3 and using RocksDB. flink-s3-fs-native uses S3TransferManager when enabled via s3.bulk-copy.enabled (default: true) for bulk copy operations and s3.async.enabled (default: true) for async read/write, providing similar performance benefits.

Back to top