> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/storage/filesystem.md).

# Filesystem

**Source** | **Sink**

The FileSystem connector lets you read from and write to object stores and local filesystems. It works with AWS S3, Google Cloud Storage, Azure Blob Storage, any S3-compatible store (MinIO, etc.), and the local filesystem. You can use it as a source (read files into a pipeline), as a sink (write pipeline output to files), or both in the same deployment.

The sink supports multiple output formats (Parquet, JSON), Hive-style partitioning, configurable file rolling policies, and multipart uploads for object stores. The source reads Parquet and JSON files with optional compression (gzip, zstd) and regex-based file filtering.

***

## Storage Backends

The `path` field determines which storage backend e6 Ingestion Engine uses. e6 Ingestion Engine parses the URI scheme to select the backend automatically - you don't need to configure the backend type separately.

* **`s3://`** - AWS S3 or any S3-compatible store. Example: `s3://my-bucket/data/`
* **`gs://`** - Google Cloud Storage. Example: `gs://my-bucket/data/`
* **`abfs://`** or **`abfss://`** - Azure Blob Storage (ADLS Gen2). Example: `abfs://container@account.dfs.core.windows.net/data/`
* **`file://`** or bare absolute paths - Local filesystem. Example: `file:///tmp/data/` or `/tmp/data/`

S3 also accepts virtual-hosted-style URLs (`https://bucket.s3.region.amazonaws.com/key`) and endpoint-embedded URLs for custom S3-compatible services (`s3::https://minio.local:9000/bucket/key`).

### Storage Options

The `storage_options` map passes authentication and configuration to the underlying `object_store` library. The keys depend on which backend the `path` resolves to.

**AWS S3:**

```yaml
storage_options:
  endpoint: https://s3.us-east-1.amazonaws.com
  region: us-east-1
  access_key_id: "${AWS_ACCESS_KEY_ID}"
  secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
```

When an `endpoint` is provided, e6 Ingestion Engine automatically sets `allow_http: true` and `virtual_hosted_style_request: false` on the S3 client. This means custom endpoints (MinIO, LocalStack, etc.) work out of the box without additional flags.

**Google Cloud Storage:**

On GKE with Workload Identity, no explicit credentials are needed. Otherwise:

```yaml
storage_options:
  gcs.project-id: my-project
  gcs.service-account-key: "${GCP_SERVICE_ACCOUNT_JSON}"
```

**Azure Blob Storage:**

```yaml
storage_options:
  azure.account-name: mystorageaccount
  azure.account-key: "${AZURE_ACCOUNT_KEY}"
```

**S3-Compatible (MinIO, etc.):**

```yaml
storage_options:
  endpoint: http://minio.example.com:9000
  region: us-east-1
  access_key_id: minioadmin
  secret_access_key: minioadmin
```

***

## Reading from the FileSystem

To read files from a storage path, configure a source with the path, a compression format, and a schema describing the file contents.

```yaml
table_type:
  type: source
  source_config:
    path: s3://my-bucket/logs/
    compression_format: gzip
    regex_pattern: ".*\\.json\\.gz$"
    storage_options:
      region: us-east-1
schema:
  format:
    json: {}
  fields:
    - field_name: timestamp
      field_type:
        type:
          primitive: UnixMicros
      nullable: false
    - field_name: message
      field_type:
        type:
          primitive: String
      nullable: false
```

### Source Config

* **`path`** (required) - URI of the folder to read from. e6 Ingestion Engine lists all objects under this prefix and reads them one by one. Supports `s3://`, `gs://`, `abfs://`, `file://`, and bare absolute paths.
* **`compression_format`** - Compression applied to the source files. `none` (default), `zstd`, or `gzip`. For JSON files, e6 Ingestion Engine wraps the byte stream in the appropriate decompression reader. Parquet files handle compression internally at the column-chunk level, so this setting only applies to JSON sources.
* **`regex_pattern`** - Regex pattern to filter which files to read. e6 Ingestion Engine lists everything under `path` recursively and only reads files whose object key matches this pattern. Optional - when omitted, all files under `path` are read.
* **`storage_options`** - Cloud storage credentials. See the Storage Options section above.

### Format

The `format` field inside `schema` determines how files are deserialized:

* **`json: {}`** - Newline-delimited JSON (one JSON object per line)
* **`parquet: {}`** - Apache Parquet (columnar format, compression handled internally)

### How the Source Works

e6 Ingestion Engine distributes files across parallel subtasks by hashing each file path and assigning it to a subtask based on the hash modulo the parallelism. Each subtask reads its assigned files sequentially. File read progress is tracked in e6 Ingestion Engine's checkpoint state - on recovery, already-finished files are skipped and partially-read files resume from the last checkpointed position.

The source is a bounded source - once all matching files have been read, the pipeline finishes. It does not watch for new files after the initial listing.

***

## Writing to the FileSystem

To write pipeline output to files, configure a sink with the destination path, a rolling policy, and a schema describing the output format.

```yaml
table_type:
  type: sink
  sink_config:
    path: s3://my-bucket/output/
    rolling_policy:
      file_size_bytes: 134217728
      interval_seconds: 300
    file_naming:
      prefix: events
      strategy: uuid
    storage_options:
      region: us-east-1
schema:
  format:
    parquet: {}
  fields:
    - field_name: event_id
      field_type:
        type:
          primitive: String
      nullable: false
```

### Sink Config

* **`path`** (required) - URI of the folder to write to. All output files land under this prefix. The path is validated at pipeline creation time - e6 Ingestion Engine parses the URI and rejects invalid schemes or malformed URLs.
* **`storage_options`** - Cloud storage credentials. Same as the source.
* **`rolling_policy`** - When to close files and start new ones. See Rolling Policy below.
* **`file_naming`** - File naming strategy and prefix/suffix. See File Naming below.
* **`partitioning`** - Hive-style data partitioning. See Partitioning below.
* **`multipart`** - Multipart upload tuning. See Multipart Upload below.

### Format

The `format` field inside `schema` determines how data is serialized to files:

* **`parquet: {}`** - Apache Parquet. Supports compression options (`snappy`, `gzip`, `zstd`, `lz4`, `uncompressed`) and configurable row group size via `row_group_bytes` (default 128 MB).
* **`json: {}`** - Newline-delimited JSON.

### Delivery Guarantee

The FileSystem sink uses a two-phase commit protocol tied to e6 Ingestion Engine's checkpoint system. On each checkpoint, in-progress multipart uploads are snapshotted (including part IDs and buffered bytes). On recovery, the sink resumes uploads from the last checkpoint - already-uploaded parts are reused and only incomplete parts are re-uploaded. Files are finalized (multipart complete) only after a successful commit, so output is exactly-once: no duplicate or partial files appear in the output path.

For local filesystem paths, the sink writes to a temporary `__in_progress` subdirectory and moves files to the final location on commit.

***

## Rolling Policy

Controls when files are closed and flushed to the object store. Multiple policies can be combined - whichever triggers first wins. A hard limit on the number of multipart parts (from `multipart.max_parts`, default 1000) is always enforced regardless of other policies.

* **`file_size_bytes`** - Close the file after it reaches this many bytes. Use this to produce predictably-sized files for downstream readers.
* **`interval_seconds`** - Close the file after this many seconds since the first write. Must be >= 1. Useful for bounding file age in high-volume pipelines.
* **`inactivity_seconds`** - Close the file after this many seconds of no new data. Must be >= 1. Good for low-volume or bursty sources where you want files flushed during idle periods.

```yaml
rolling_policy:
  file_size_bytes: 536870912    # 512 MB
  interval_seconds: 300          # 5 minutes
  inactivity_seconds: 10         # 10 seconds idle
```

For high-volume sources, combine `file_size_bytes` and `interval_seconds` to get both size-bounded and time-bounded files. For low-volume or bursty sources, `inactivity_seconds` alone is often sufficient.

**Important:** At least one rolling policy should be configured. Without it, files are only rolled when the multipart part limit (default 1000 parts) is reached, which can produce very large files.

***

## File Naming

Controls how output files are named. The final filename follows the pattern `{prefix}-{id}.{suffix}` (or `{id}.{suffix}` if no prefix is set).

* **`prefix`** - String prepended to the filename. Example: with prefix `events`, files are named `events-<id>.parquet`.
* **`suffix`** - Overrides the default file extension. Defaults to the format's natural extension (`.parquet` for Parquet, `.json` for JSON). Use with caution - changing the suffix does not change the actual format.
* **`strategy`** - How the `{id}` portion is generated. Default: `uuid`.

### Naming Strategies

| Strategy         | Output example                                        |
| ---------------- | ----------------------------------------------------- |
| `uuid` (default) | `events-1f2d6a31-8266-4ba7-9f1f-5d3a9fab3b86.parquet` |
| `uuid_v7`        | `events-01912345-abcd-7000-8000-000000000000.parquet` |
| `ulid`           | `events-01HZ3KPBV4QKJM8X9YGDNR5KC.parquet`            |
| `serial`         | `events-00001-000.parquet`                            |

The `serial` strategy produces filenames like `{prefix}-{file_index}-{subtask_index}.{suffix}`, where the file index increments each time a file is rolled and the subtask index identifies the parallel writer. The other strategies generate a globally unique ID per file.

```yaml
file_naming:
  prefix: events
  strategy: uuid_v7
```

***

## Hive-Style Partitioning

Partition output files into directories based on column values. Produces a directory layout like `country=US/events-17761234.parquet`.

### Partition Fields

* **`partitioning.fields`** - List of columns to partition by. Each entry has:
  * **`name`** - Column name. Supports dotted paths for nested structs (e.g., `geo.country`).
  * **`transform`** - Transform to apply before partitioning. Default: `identity`.

### Transforms

| Transform            | Output directory            | Applies to                     |
| -------------------- | --------------------------- | ------------------------------ |
| `identity` (default) | `field=value/`              | Any type                       |
| `hour`               | `field_hour=<epoch_hours>/` | Timestamp/DateTime fields only |
| `year`               | `field_year=YYYY/`          | Timestamp/DateTime fields only |
| `month`              | `field_month=M/`            | Timestamp/DateTime fields only |

The `hour`, `year`, and `month` transforms only work with timestamp or date fields (`UnixMicros`, `UnixMillis`, `UnixNanos`, `DateTime`). Using them on non-temporal fields produces a validation error at pipeline creation time.

### Time Pattern Rolling

* **`partitioning.time_pattern`** - A date/time format string (e.g., `%Y/%m/%d`). When set, the sink uses watermark-based rolling: a file is rolled when the pipeline watermark advances past the time bucket of the file's first record. This ensures files align with time boundaries without relying on wall-clock timers.

### Partition Shuffle

* **`partitioning.shuffle_by_partition.enabled`** - When `true`, e6 Ingestion Engine hash-shuffles records by partition keys before they reach the sink. This concentrates all records for a given partition value onto a single subtask, reducing the total number of output files. Default: `false`. Enable this for pipelines with many subtasks writing to the same partition keys. Disable it (or leave default) for single-writer pipelines or when data is heavily skewed - shuffling skewed data can cause backpressure on the hot subtask.

The partition columns must exist in the pipeline SQL's SELECT projection and in the sink table's schema fields. e6 Ingestion Engine validates partition fields against the schema at pipeline creation time.

```yaml
partitioning:
  fields:
    - name: country
      transform: identity
    - name: event_time
      transform: hour
  time_pattern: "%Y-%m-%d-%H"
  shuffle_by_partition:
    enabled: false
```

Resulting layout:

```
s3://my-bucket/output/
├── country=US/event_time_hour=488689/events-abc123.parquet
├── country=US/event_time_hour=488690/events-def456.parquet
├── country=UK/event_time_hour=488689/events-ghi789.parquet
└── ...
```

***

## Multipart Upload

Tuning for object store multipart uploads. These settings affect how the sink splits data into parts for upload. Usually not needed - the defaults work for most cases.

* **`target_part_size_bytes`** - Target size for each part of the multipart upload, in bytes. Default: 33554432 (32 MB). Minimum: 5242880 (5 MB) - the S3 minimum part size. e6 Ingestion Engine enforces this minimum and rejects smaller values at pipeline creation time.
* **`max_parts`** - Maximum number of parts per upload. Default: 1000. This acts as a hard rolling policy - when a file reaches this many parts, it is rolled regardless of other rolling policy settings.

```yaml
multipart:
  target_part_size_bytes: 67108864   # 64 MB per part
  max_parts: 500
```

***

## Complete Example

A full configuration reading gzipped JSON log files from S3 and writing processed output as Parquet, partitioned by region.

**Source** - read log files matching a regex pattern:

```yaml
table_type:
  type: source
  source_config:
    path: s3://my-bucket/raw-logs/
    compression_format: gzip
    regex_pattern: ".*\\.json\\.gz$"
    storage_options:
      region: us-east-1
      access_key_id: "${AWS_ACCESS_KEY_ID}"
      secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
schema:
  format:
    json: {}
  fields:
    - field_name: event_id
      field_type:
        type:
          primitive: String
      nullable: false
    - field_name: region
      field_type:
        type:
          primitive: String
      nullable: false
    - field_name: event_time
      field_type:
        type:
          primitive: UnixMicros
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: String
      nullable: true
```

**Sink** - write Parquet files partitioned by region, rolling every 512 MB or 5 minutes:

```yaml
table_type:
  type: sink
  sink_config:
    path: s3://my-bucket/processed/
    rolling_policy:
      file_size_bytes: 536870912
      interval_seconds: 300
    file_naming:
      prefix: events
      strategy: uuid_v7
    partitioning:
      fields:
        - name: region
          transform: identity
      shuffle_by_partition:
        enabled: true
    storage_options:
      region: us-east-1
      access_key_id: "${AWS_ACCESS_KEY_ID}"
      secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
schema:
  format:
    parquet: {}
  fields:
    - field_name: event_id
      field_type:
        type:
          primitive: String
      nullable: false
    - field_name: region
      field_type:
        type:
          primitive: String
      nullable: false
    - field_name: event_time
      field_type:
        type:
          primitive: UnixMicros
      nullable: false
```

***

## JSON Schema Reference

<details>

<summary>Source Table Schema</summary>

```json
{
  "type": "object",
  "properties": {
    "table_type": {
      "type": "object",
      "properties": {
        "type": { "const": "source" },
        "source_config": {
          "type": "object",
          "properties": {
            "path": { "type": "string", "description": "URI of the folder to read from" },
            "compression_format": {
              "type": "string",
              "enum": ["none", "zstd", "gzip"],
              "default": "none"
            },
            "regex_pattern": { "type": "string" },
            "storage_options": {
              "type": "object",
              "additionalProperties": { "type": "string" }
            }
          },
          "required": ["path"]
        }
      }
    }
  }
}
```

</details>

<details>

<summary>Sink Table Schema</summary>

```json
{
  "type": "object",
  "properties": {
    "table_type": {
      "type": "object",
      "properties": {
        "type": { "const": "sink" },
        "sink_config": {
          "type": "object",
          "properties": {
            "path": { "type": "string", "description": "URI of the folder to write to" },
            "storage_options": {
              "type": "object",
              "additionalProperties": { "type": "string" }
            },
            "rolling_policy": {
              "type": "object",
              "properties": {
                "file_size_bytes": { "type": "integer", "description": "Roll after this many bytes" },
                "interval_seconds": { "type": "integer", "minimum": 1, "description": "Roll after this many seconds" },
                "inactivity_seconds": { "type": "integer", "minimum": 1, "description": "Roll after this many idle seconds" }
              }
            },
            "file_naming": {
              "type": "object",
              "properties": {
                "prefix": { "type": "string" },
                "suffix": { "type": "string" },
                "strategy": {
                  "type": "string",
                  "enum": ["uuid", "uuid_v7", "ulid", "serial"],
                  "default": "uuid"
                }
              }
            },
            "partitioning": {
              "type": "object",
              "properties": {
                "time_pattern": { "type": "string", "description": "Date/time format for watermark-based rolling" },
                "fields": {
                  "type": "array",
                  "items": {
                    "type": "object",
                    "properties": {
                      "name": { "type": "string" },
                      "transform": {
                        "type": "string",
                        "enum": ["identity", "hour", "year", "month"],
                        "default": "identity"
                      }
                    },
                    "required": ["name"]
                  }
                },
                "shuffle_by_partition": {
                  "type": "object",
                  "properties": {
                    "enabled": { "type": "boolean", "default": false }
                  }
                }
              }
            },
            "multipart": {
              "type": "object",
              "properties": {
                "target_part_size_bytes": {
                  "type": "integer",
                  "minimum": 5242880,
                  "default": 33554432,
                  "description": "Target part size (min 5 MB, default 32 MB)"
                },
                "max_parts": {
                  "type": "integer",
                  "default": 1000,
                  "description": "Max parts per upload (also acts as a hard rolling limit)"
                }
              }
            }
          },
          "required": ["path"]
        }
      }
    }
  }
}
```

</details>


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.e6data.com/ingestion-engine/connectors/storage/filesystem.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.