FS Source Node

The fssource node reads files from Amazon S3, Google Cloud Storage, Azure Blob Storage, an SFTP server, or the local filesystem, and emits their deserialized contents as pipeline records. A single backend field selects the storage system, so the same node type covers every backend without changing the rest of the pipeline. It is a batch source: each run lists the files under root, processes the ones it has not already completed, and then terminates.

Key Features

• Five backends: Amazon S3 (s3), Google Cloud Storage (gs), Azure Blob Storage (azblob), SFTP (sftp), and the local filesystem (file)
• Batch runs that resume: each run processes only files not yet completed; progress is checkpointed per file, so a rerun or a scheduled run does not re-emit records already emitted
• Encoding-driven parsing: file contents are deserialized with the configured encodingType (CSV, JSON, JSON Lines, text, XML)
• Gzip auto-detection: gzipped files are detected from their magic bytes and decompressed as they are read; no configuration needed
• Bounded memory: line-delimited and record-sequence formats are streamed in chunks; a per-file cap (maxFileBytes) protects the process from oversized payloads
• File name filtering and ordering: a regex filters file names, and an optional (?<ts>...) capture group orders files by the timestamp in their name
• Per-run cap: maxFilesPerRun bounds how much one run takes on; the oldest files go first and the rest follow on the next run
• Single-object pinning (S3): exactObjectKey restricts a run to one object below the root
• Replica sharding: when a job runs as several replicas, files are divided between them by consistent hash so no file is read twice
• Dead-lettering: records that fail to deserialize are written to the DLQ (when configured) with their source file, while the rest of the file is still emitted

Configuration

Field	Type	Required	Default	Description
backend	String	Yes	—	Storage backend. One of s3, gs, azblob, sftp, file
root	String	Yes	—	Root URI or path to list. Format depends on the backend (see Backends). Every file below it, including files in nested folders, is a candidate
encodingType	String	Yes	—	How file contents are deserialized. One of CSV, JSON_OBJECT, JSON_ARRAY, JSON_OBJECT_LINE, STRING_LINE, TEXT, XML. See Encoding Types. PARQUET is not supported by this source
fileNameRegex	String	No	—	Java regex matched against the file name (not the full path). Only matching files are processed. A named capture group (?<ts>...) sets the file's ordering timestamp (see File Name Regex and Timestamp Ordering)
exactObjectKey	String	No	—	s3 only. Restricts the run to this one object key, which must be below the root prefix. A key, not a URI, e.g. incoming/2026/09/events.jsonl
maxFileBytes	long	No	268435456 (256 MiB)	Cap, in decompressed bytes, on what is held in memory at once: the whole file for JSON_OBJECT, TEXT, and XML; a single line for JSON_OBJECT_LINE and STRING_LINE; a single record for CSV and JSON_ARRAY. A file that exceeds it is dropped (see Failure Handling). Must be greater than 0
chunkSizeBytes	int	No	16777216 (16 MiB)	Target size, in decompressed bytes, of one chunk read and emitted at a time for line-delimited and record-sequence formats. Must be greater than 0
maxFilesPerRun	int	No	10000	Maximum number of files one run processes. When more are pending, the oldest are taken and the rest are left for the next run. Must be greater than 0
backendConfig	Map<String, Object>	No	—	Backend-specific credentials and settings. See Backends

Backends

The backend field selects the storage system. Backend-specific credentials and settings go in the backendConfig map. Credentials are referenced by ID from jobContext.otherProperties, never inlined.

s3 — Amazon S3

Reads objects from an S3 bucket or an S3-compatible store. root is an S3 URI: s3://bucket for the whole bucket or s3://bucket/prefix for a folder. The prefix is treated as a folder, so s3://my-bucket/logs matches logs/... but not logs-archive/.... All objects below the prefix are listed, including those under nested prefixes.

backendConfig key	Required	Default	Description
region	No	us-east-1	AWS region of the bucket
credentialId	No	—	ID of a username/password credential in jobContext.otherProperties, where username is the AWS access key ID and password is the secret access key. Omit it to use the AWS SDK default credential chain (environment variables, instance profile, or IAM role). If it is set but cannot be found, the node fails at startup
s3EndpointOverride	No	—	Custom endpoint URL for S3-compatible storage (e.g. MinIO, LocalStack). Setting it also switches the client to path-style addressing

jobContext:
  otherProperties:
    aws-cred:
      username: AKIAIOSFODNN7EXAMPLE
      password: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

dag:
  - id: source
    commandName: fssource
    config:
      backend: s3
      root: s3://my-bucket/incoming/
      encodingType: JSON_OBJECT_LINE
      backendConfig:
        region: us-east-1
        credentialId: aws-cred

IAM permissions

The AWS identity must be able to list the bucket and read its objects:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::my-bucket"
    },
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::my-bucket/*"
    }
  ]
}

s3:ListBucket can be narrowed to the root prefix with an s3:prefix condition. The source never writes to, moves, or deletes objects in the bucket.

Reading a single object

Set exactObjectKey to process one object instead of everything under root. The key must sit below the root prefix: with root: s3://my-bucket/incoming/, the key incoming/2026/09/30/events.jsonl is accepted, while other/events.jsonl or s3://my-bucket/incoming/events.jsonl is rejected at validation. fileNameRegex still applies to the pinned object. exactObjectKey is not supported on any other backend.

gs — Google Cloud Storage

Reads objects from a GCS bucket. root is a GCS URI: gs://bucket or gs://bucket/prefix.

backendConfig key	Required	Default	Description
serviceAccountJson	No	—	Full service account JSON key as a string. Omit to use Application Default Credentials

config:
  backend: gs
  root: gs://my-bucket/incoming/
  encodingType: JSON_OBJECT_LINE
  backendConfig:
    serviceAccountJson: |
      {
        "type": "service_account",
        "project_id": "my-project",
        "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n",
        "client_email": "reader@my-project.iam.gserviceaccount.com"
      }

azblob — Azure Blob Storage

Reads blobs from an Azure Storage container. root is the container URL, optionally with a prefix: https://<account>.blob.core.windows.net/<container>/<prefix>.

Provide exactly one form of authentication:

backendConfig key	Required	Default	Description
connectionString	Conditional	—	Full Azure Storage connection string. Takes precedence when both keys are set
credentialId	Conditional	—	ID of a username/password credential in jobContext.otherProperties, where username is the storage account name and password is the account key

config:
  backend: azblob
  root: https://mystorage.blob.core.windows.net/logs/incoming/
  encodingType: JSON_OBJECT_LINE
  backendConfig:
    credentialId: azure-storage-cred

sftp — SFTP

Reads files from an SFTP server. root is sftp://host[:port]/path; the port defaults to 22.

Provide exactly one of credentialId or privateKeyCredentialId:

backendConfig key	Required	Default	Description
credentialId	Conditional	—	ID of a username/password credential in jobContext.otherProperties for password authentication
privateKeyCredentialId	Conditional	—	ID of an RSA private key credential in jobContext.otherProperties for key authentication
hostKeyFingerprint	No	—	Expected server host key fingerprint. When omitted, the server's host key is not verified; set it in production

config:
  backend: sftp
  root: sftp://sftp.example.com:22/exports/daily
  encodingType: CSV
  backendConfig:
    credentialId: sftp-cred
    hostKeyFingerprint: "SHA256:AbCdEf..."

file — Local Filesystem

Reads files from the filesystem of the host running the pipeline. root is an absolute directory path or a file: URI. No backendConfig is required.

config:
  backend: file
  root: /data/input
  encodingType: STRING_LINE

How a Run Works

fssource is a batch source. Each execution performs a single pass and then terminates. To pick up new files, run the pipeline again, for example on a schedule.

1. List: every file below root is listed, filtered by fileNameRegex and, on S3, exactObjectKey.
2. Skip completed files: files already recorded in the checkpoint, or older than the checkpoint watermark, are dropped.
3. Select: the remaining files are sorted oldest first, and at most maxFilesPerRun are taken.
4. Read and emit: each file is opened, gunzipped if needed, deserialized with encodingType, and emitted downstream. Line-delimited formats are emitted one chunk at a time, and CSV and JSON arrays are streamed record by record in batches of about chunkSizeBytes.
5. Checkpoint: once a file is fully handled, it is recorded as completed and the watermark advances.

If a listing matches no objects at all, the node logs a warning naming the root, so check the prefix and filters.

Checkpointing and resume

Checkpoints are stored through the checkpoint service named by the CHECKPOINT_URL property in jobContext.otherProperties. The Fleak scheduler sets this, together with a CHECKPOINT_SCOPE that keeps each workflow's progress separate even when several workflows read the same folder.

• Without CHECKPOINT_URL, checkpoints are held in memory only, so every run re-reads all matching files.
• Without CHECKPOINT_SCOPE, progress is scoped to the current job ID, so a new job does not inherit an earlier job's progress.

Each checkpoint tracks a watermark (the ordering timestamp of processed files) and the set of files completed since then. A file whose ordering timestamp is older than the watermark is skipped even if it was never read. Files written late with an old lastModified, or with an old timestamp in their name, are therefore not picked up. A file that could not be read holds its shard's watermark at its timestamp, so it is retried on the next run instead of being skipped.

Replica sharding

When a job runs with more than one replica, the REPLICA_INDEX and REPLICA_COUNT properties in jobContext.otherProperties divide the files between replicas. Each file is assigned by a consistent hash of its URI, so the assignment is stable across runs and every file belongs to exactly one replica. Without these properties, a single instance processes all files.

File Name Regex and Timestamp Ordering

fileNameRegex is a Java regex matched against the file name only (the part after the last /), and it must match the whole name. Files whose names do not match are skipped entirely.

If the regex contains a named capture group (?<ts>...), the captured value is parsed as Unix epoch seconds and used as the file's ordering timestamp. Files are processed in ascending timestamp order, which matters when downstream logic depends on the order in which files were produced.

fileNameRegex: "events-(?<ts>[0-9]+)\\.jsonl(\\.gz)?"

Without a ts group, or when the captured value is not a number, the file's lastModified timestamp reported by the backend is used instead.

Failure Handling

Situation	Behavior
File exceeds maxFileBytes	Dropped and checkpointed as done, so it is not retried. Records already emitted from it before the cap was hit stand. Counted in input_file_skipped_count with skip_reason=file_too_large
File cannot be read (e.g. permission denied, network error, corrupt gzip)	Skipped for this run and retried on the next. skip_reason=read_error
Some records fail to deserialize	The good records are emitted. The bad ones are counted in input_deser_err_count and written to the DLQ with the source file URI (__md_fs_source_urn) and record index (__md_fs_source_record_index) as DLQ metadata. The file is marked completed
No record in a file deserializes and there is no DLQ	Skipped and retried on the next run. skip_reason=nothing_deserialized
A downstream node fails while accepting records	The file is skipped and retried on the next run. skip_reason=downstream_error

If a run emits no file and at least one file was skipped for retry, the run fails instead of reporting success, so the scheduler does not treat an unread batch as done. Files dropped for exceeding maxFileBytes do not count as skipped.

Dead-lettering requires a dlqConfig on the job context. Without one, malformed records are logged and counted but not stored.

DAG Examples

Batch ingest from S3

Reads every .jsonl or .jsonl.gz object under incoming/ that has not already been processed, and forwards each line as a record.

jobContext:
  otherProperties:
    aws-cred:
      username: AKIAIOSFODNN7EXAMPLE
      password: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
  metricTags: {}
  dlqConfig:

dag:
  - id: source
    commandName: fssource
    config:
      backend: s3
      root: s3://my-bucket/incoming/
      fileNameRegex: ".*\\.jsonl(\\.gz)?"
      encodingType: JSON_OBJECT_LINE
      maxFilesPerRun: 500
      backendConfig:
        region: us-east-1
        credentialId: aws-cred
    outputs:
      - sink

  - id: sink
    commandName: stdout
    config:
      encodingType: JSON_OBJECT

Single object from an S3-compatible store

Reads one CSV object from a MinIO bucket, using the default credential chain.

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: source
    commandName: fssource
    config:
      backend: s3
      root: s3://exports/daily/
      exactObjectKey: daily/2026-09-30.csv
      encodingType: CSV
      backendConfig:
        region: us-east-1
        s3EndpointOverride: http://minio.internal:9000
    outputs:
      - sink

  - id: sink
    commandName: stdout
    config:
      encodingType: JSON_OBJECT

CSV from GCS in file-name timestamp order

Processes report-<epochSeconds>.csv files in the order given by the timestamp in their names.

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: source
    commandName: fssource
    config:
      backend: gs
      root: gs://my-bucket/reports/
      fileNameRegex: "report-(?<ts>[0-9]+)\\.csv"
      encodingType: CSV
    outputs:
      - sink

  - id: sink
    commandName: stdout
    config:
      encodingType: JSON_OBJECT

Related Nodes

• s3rtsource: Stream new S3 objects as they arrive, driven by S3 event notifications through SQS
• s3sink: Write records to Amazon S3
• gcssink: Write records to a Google Cloud Storage bucket
• azureblobsink: Write records to Azure Blob Storage