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 Ingest Engine uses. Ingest 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://[email protected]/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:

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, Ingest 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:

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

Azure Blob Storage:

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

S3-Compatible (MinIO, etc.):

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.

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. Ingest 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, Ingest 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. Ingest 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

Ingest 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 Ingest 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.

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 - Ingest 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 Ingest 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.

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.

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, Ingest 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. Ingest Engine validates partition fields against the schema at pipeline creation time.

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. Ingest 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.

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:

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:

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

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Storage Backends

Storage Options

Reading from the FileSystem

Source Config

Format

How the Source Works

Writing to the FileSystem

Sink Config

Format

Delivery Guarantee

Rolling Policy

File Naming

Naming Strategies

Hive-Style Partitioning

Partition Fields

Transforms

Time Pattern Rolling

Partition Shuffle

Multipart Upload

Complete Example

JSON Schema Reference

​Storage Backends

​Storage Options

​Reading from the FileSystem

​Source Config

​Format

​How the Source Works

​Writing to the FileSystem

​Sink Config

​Format

​Delivery Guarantee

​Rolling Policy

​File Naming

​Naming Strategies

​Hive-Style Partitioning

​Partition Fields

​Transforms

​Time Pattern Rolling

​Partition Shuffle

​Multipart Upload

​Complete Example

​JSON Schema Reference

Storage Backends

Storage Options

Reading from the FileSystem

Source Config

Format

How the Source Works

Writing to the FileSystem

Sink Config

Format

Delivery Guarantee

Rolling Policy

File Naming

Naming Strategies

Hive-Style Partitioning

Partition Fields

Transforms

Time Pattern Rolling

Partition Shuffle

Multipart Upload

Complete Example

JSON Schema Reference