# S3 input connector

:::note
This page describes configuration options specific to the S3 input connector.
See [top-level connector documentation](/connectors/) for general information
about configuring input and output connectors.
:::

The S3 input connector is used to load data from an S3 bucket to a Feldera table.
It can be configured to load a single object or multiple objects selected based on a
common S3 prefix. By setting the `endpoint_url`, you can also read from non-AWS services
that offer S3 compatible APIs.

:::tip
When accessing an S3 bucket that stores data in the Delta Lake or Iceberg format, consider
using the [Delta Lake connector](/connectors/sources/delta) or the
[Iceberg connector](/connectors/sources/iceberg) connector instead.
:::

The S3 input connector supports [fault tolerance](/pipelines/fault-tolerance).

## Configuration options

| Property                      | Type   | Default    | Description   |
|-------------------------------|--------|------------|---------------|
| `aws_access_key_id`           | string |            | AWS Access Key id. This property must be specified unless `no_sign_request` is set to `true`.|
| `aws_secret_access_key`       | string |            | Secret Access Key. This property must be specified unless `no_sign_request` is set to `true`.|
| `no_sign_request`             | bool   | `false`    | Do not sign requests. This is equivalent to the `--no-sign-request` flag in the AWS CLI. |
| `key`[^1]                     | string |            | Read a single object specified by a key. |
| `prefix`[^1]                  | string |            | Read all objects whose keys match a prefix. Set to an empty string to read all objects in the bucket. |
| `region`[^2]                  | string |            | AWS region. |
| `bucket_name`[^2]             | string |            | S3 bucket name. |
| `endpoint_url`                | string |            | The endpoint URL used to communicate with this service. Explicitly set it to connect to non-AWS services. For example, use `https://storage.googleapis.com` to interact with Google Cloud Storage. |
| `max_concurrent_fetches`      | integer| 8          | <p>Controls the number of S3 objects fetched in parallel.</p><p>Increasing this value can improve throughput by enabling greater concurrency. However, higher concurrency may lead to timeouts or increased memory usage due to in-memory buffering.</p><p>Recommended range: 1–10. Default: 8.</p> |
| `max_retries`                 | integer| 5          | Retry `max_retries` times with exponential backoff. <p>If the object changes during a retry attempt, the remaining part of the object will not be processed.</p> |

:::tip
When `prefix` is used, Feldera reads objects in alphabetical order,
the order returned by S3's [ListObjectsV2] operation.  When
`max_concurrent_fetches` is greater than 1, data is processed in
parallel for up to the specified number of objects, rather than
sequentially from one object to the next.
:::

[^1]: Exactly one of `key` or `prefix` must be set.
[^2]: This property is required.

[ListObjectsV2]: https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html

### Support for AWS IAM roles for service accounts (IRSA)

To use [AWS IAM roles for service accounts (IRSA)](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html),
you must **skip** supplying the fields `aws_access_key_id` and `aws_secret_access_key`.

Feldera will pick up the credentials from the environment variables.

## Examples

Populate a table from a JSON file in a public S3 bucket:

```sql
CREATE TABLE vendor (
    id BIGINT NOT NULL PRIMARY KEY,
    name VARCHAR,
    address VARCHAR
) WITH ('connectors' = '[{
    "transport": {
        "name": "s3_input",
        "config": {
            "key": "vendor.json",
            "no_sign_request": true,
            "bucket_name": "feldera-basics-tutorial",
            "region": "us-west-1"
        }
    },
    "format": { "name": "json" }
}]');
```

Populate a table from a JSON file, using access key-based authentication:

```sql
CREATE TABLE vendor (
    id BIGINT NOT NULL PRIMARY KEY,
    name VARCHAR,
    address VARCHAR
) WITH ('connectors' = '[{
    "transport": {
        "name": "s3_input",
        "config": {
            "key": "vendor.json",
            "aws_access_key_id": "YOUR_ACCESS_KEY_ID",
            "aws_secret_access_key": "YOUR_SECRET_ACCESS_KEY",
            "bucket_name": "feldera-basics-tutorial",
            "region": "us-west-1"
        }
    },
    "format": { "name": "json" }
}]');
```

To connect to Google Cloud Storage, [explicitly set](https://cloud.google.com/storage/docs/interoperability) the `endpoint_url`. You also
need to configure an [HMAC
key](https://cloud.google.com/storage/docs/authentication/hmackeys) to get an
Access ID and Secret. You will also need to grant the corresponding Principal
access to the bucket.

```sql
CREATE TABLE vendor (
    id BIGINT NOT NULL PRIMARY KEY,
    name VARCHAR,
    address VARCHAR
) WITH ('connectors' = '[{
    "transport": {
        "name": "s3_input",
        "config": {
            "key": "vendor.json",
            "aws_access_key_id": "YOUR_ACCESS_KEY_ID",
            "aws_secret_access_key": "YOUR_SECRET_ACCESS_KEY",
            "bucket_name": "my-bucket",
            "region": "us-west1",
            "endpoint_url": "https://storage.googleapis.com"
        }
    },
    "format": { "name": "json" }
}]');
```

:::tip
Refer to the [secret references](/connectors/secret-references) guide
to externalize AWS access keys via Kubernetes.
:::


## Additional resources

For more information, see:

* [Top-level connector documentation](/connectors/)
* [Supported data formats](/formats)
* [Delta Lake input connector documentation](/connectors/sources/delta)
