Skip to main content

FS Source Node

The fssource node reads files from a local filesystem, Amazon S3, or Google Cloud Storage and emits their content as pipeline records. A single configuration controls which backend is used, so the same node type covers local development, AWS, and GCP without changing anything else in the pipeline. It supports both batch (read a fixed set of files and stop) and streaming (continuously poll for new files) modes.

Key Features​

  • Three backends: local filesystem (file), AWS S3 (s3), and Google Cloud Storage (gs) — selected by a single backend field
  • Batch and streaming modes: BOUNDED reads existing files once and terminates; UNBOUNDED polls continuously for new files
  • Three emission types: emit one record per line (LINE), one record per file (WHOLE_FILE), or file metadata without reading content (FILE_REFERENCE)
  • Post-actions: after a file is consumed it can be left alone (NONE), deleted (DELETE), or moved to an archive prefix (ARCHIVE)
  • Stability probe: optionally hold a file until its size stops changing, guarding against partially-written files
  • Checkpointing: progress is persisted to _zephflow_checkpoints/ so restarts do not reprocess already-consumed files
  • Parallel partitioning: multiple job instances divide the file set by consistent hash so no file is processed twice

Configuration​

FieldTypeRequiredDefaultDescription
backendStringYes—Storage backend. One of file, s3, gs
rootStringYes—Root path or URI to scan. Format depends on the backend (see Backends)
fileNameRegexStringNo—Java regex applied to the file name (not the full path). Only matching files are processed. Supports a named capture group (?<ts>...) for timestamp-based ordering
modeStringNoBOUNDEDBOUNDED to read existing files once and stop; UNBOUNDED to poll continuously for new files
listingIntervalMslongNo30000Milliseconds to wait between listing passes. In UNBOUNDED mode this is the delay after a pass that found files; backoff applies when no files are found
emissionEmissionNo{type: LINE}Controls how file contents are turned into records. See Emission Types
stabilityStabilityNo{enabled: false}Stability probe configuration. See Stability Probe
postActionPostActionConfigNo{type: NONE}What to do with a file after it is consumed. See Post-Actions
partitionPartitionConfigNo—Explicit partition assignment for parallel execution. See Partitioning
checkpointCheckpointOverrideNo—Override where checkpoints are stored. See Checkpointing
backendConfigMap<String, Object>No—Backend-specific credentials and settings. See Backends

Emission Object​

FieldTypeRequiredDefaultDescription
typeStringYesLINELINE, WHOLE_FILE, or FILE_REFERENCE
encodingStringNoutf-8Character encoding used when reading file bytes
lineBatchSizeintNo500Number of lines to accumulate before flushing downstream (only applies to LINE mode)

Stability Object​

FieldTypeRequiredDefaultDescription
enabledbooleanYesfalseEnable the size-stability check before reading
probeDelayMslongNo10000How long to wait after first seeing a file before checking whether its size has stabilised

PostActionConfig Object​

FieldTypeRequiredDefaultDescription
typeStringYesNONENONE, DELETE, or ARCHIVE
destinationPrefixStringConditional—Required when type is ARCHIVE. Target URI prefix files are moved to

PartitionConfig Object​

FieldTypeRequiredDefaultDescription
parallelismintYes—Total number of parallel job instances sharing this source
indexintYes—Zero-based index of this job instance. Must be in [0, parallelism)

CheckpointOverride Object​

FieldTypeRequiredDefaultDescription
backendStringYes—Backend to use for checkpoint storage (file, s3, or gs)
rootStringYes—Root URI for the checkpoint store

Backends​

The backend field selects which storage system the node reads from. Backend-specific credentials and settings are passed as a backendConfig map.

file — Local Filesystem​

Reads files from the local filesystem of the host running the pipeline. root is an absolute directory path, and no backendConfig is required.

config:
  backend: file
  root: /data/input

s3 — Amazon S3​

Reads objects from an S3 bucket. root is an S3 URI (s3://bucket/prefix).

Credentials are resolved by the AWS SDK default credential chain — environment variables (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY), instance profile, or IAM role.

backendConfig keyRequiredDefaultDescription
regionNous-east-1AWS region of the bucket
s3EndpointOverrideNo—Custom endpoint URL for S3-compatible storage (e.g. MinIO)
config:
  backend: s3
  root: s3://my-bucket/incoming/
  backendConfig:
    region: us-east-1

gs — Google Cloud Storage​

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

backendConfig keyRequiredDefaultDescription
serviceAccountJsonNo—Full service account JSON key as a string. Omit to use Application Default Credentials
config:
  backend: gs
  root: gs://my-bucket/incoming/
  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"
      }

Emission Types​

The emission.type field controls how each file's bytes are turned into pipeline records.

TypeRecord fields emittedDescription
LINEline, fileOne record per line of text. line is the text of the line; file is the full URN of the source file
WHOLE_FILEfile, contentOne record per file. content holds the entire decoded file text
FILE_REFERENCEfile, size, lastModified, displayPathOne record per file containing only metadata — the file is never read. Useful for triggering downstream processing

LINE example output:

{"line": "{\"id\":1,\"event\":\"login\"}", "file": "s3://my-bucket/incoming/events.jsonl"}

WHOLE_FILE example output:

{"file": "gs://my-bucket/reports/summary.txt", "content": "Total: 42\nErrors: 0\n"}

FILE_REFERENCE example output:

{
  "file": "s3://my-bucket/incoming/events.jsonl",
  "size": 204800,
  "lastModified": "2026-06-05T13:00:00Z",
  "displayPath": "s3://my-bucket/incoming/events.jsonl"
}

Operating Modes​

BOUNDED (batch)​

The node lists all matching files once, processes them in order, and calls terminate() when the list is exhausted. If no files match on the first pass the node terminates immediately with no records emitted.

Use this mode for one-shot backfills, replay jobs, or any pipeline where the input set is known in advance.

UNBOUNDED (streaming)​

The node polls the root path repeatedly. After each pass that found and processed files it waits listingIntervalMs before listing again. When a pass finds nothing new, it applies an exponential backoff starting at 100 ms, capped at listingIntervalMs. This avoids hammering the storage API during quiet periods.

Files already recorded in the checkpoint are skipped on every pass, so each file is processed exactly once across restarts.

Post-Actions​

A post-action runs on each file after it has been fully emitted and checkpointed.

typeBehaviour
NONEFile is left untouched (default)
DELETEFile is deleted from the backend
ARCHIVEFile is moved (copy + delete) to postAction.destinationPrefix. The filename is preserved; only the containing prefix changes

ARCHIVE example — move processed S3 objects to a processed/ prefix in the same bucket:

postAction:
  type: ARCHIVE
  destinationPrefix: s3://my-bucket/processed/

Stability Probe​

When stability.enabled is true, the node will not read a file until its size has been unchanged for at least probeDelayMs milliseconds. On first sight the file is skipped and a baseline is recorded. On subsequent listing passes the file is re-statted; if the size and last-modified timestamp match the baseline after the delay has elapsed, the file is considered stable and processed normally.

This is useful when files are written incrementally (e.g. log rotation in progress) and you want to avoid reading a partially-written file.

stability:
  enabled: true
  probeDelayMs: 15000

Checkpointing​

Progress is stored in a _zephflow_checkpoints/ folder under the root URI. Each checkpoint tracks a watermark (the timestamp of the latest-seen file) and the set of URNs that have been fully processed. On restart the node loads its checkpoint, skips all already-processed files, and resumes from where it left off.

The checkpoint key is derived from a stable hash of (backend, root, fileNameRegex) combined with the partition index and parallelism, so different pipeline configurations and different shards never collide.

Override checkpoint location​

By default checkpoints are stored alongside the source data. Use checkpoint to write them to a different backend or path — for example, storing GCS source checkpoints in a dedicated S3 bucket:

checkpoint:
  backend: s3
  root: s3://checkpoints-bucket/fssource/

Partitioning​

When multiple job instances run the same pipeline in parallel, each instance must process a disjoint subset of files. The partition config makes this explicit:

partition:
  parallelism: 4
  index: 0       # this instance handles shard 0 of 4

Files are assigned to shards by consistent hashing of the file URN, so the assignment is stable across restarts and listing passes. Each file is owned by exactly one shard.

If partition is omitted, the node reads zephflow.job.parallelism and zephflow.job.index from jobContext.otherProperties. If those are also absent it defaults to parallelism=1, index=0 (all files processed by a single instance).

File Name Regex and Timestamp Ordering​

fileNameRegex is a Java regex matched against the file name only (not the full path). Files whose names do not match are skipped entirely.

If the regex contains a named capture group (?<ts>\d+), the captured value is parsed as a Unix epoch second and used as the file's sort key. Files are processed in ascending timestamp order, which is important for UNBOUNDED pipelines where you want to process files in the order they were produced.

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

Without a ts group, files are sorted by their lastModified timestamp reported by the backend, with the URN as a tiebreaker.

DAG Examples​

Batch ingest from S3 (line-by-line)​

Reads every .jsonl file under incoming/ once and forwards each line as a record.

dag:
  - id: source
    commandName: fssource
    config:
      backend: s3
      root: s3://my-bucket/incoming/
      fileNameRegex: ".*\\.jsonl"
      mode: BOUNDED
      emission:
        type: LINE
      backendConfig:
        region: us-east-1
    outputs:
      - sink

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

Streaming ingest from GCS (whole-file, with archiving)​

Polls reports/ every 60 seconds in streaming mode, reads each new file as a single record, and moves processed files to archive/.

dag:
  - id: source
    commandName: fssource
    config:
      backend: gs
      root: gs://my-bucket/reports/
      mode: UNBOUNDED
      listingIntervalMs: 60000
      emission:
        type: WHOLE_FILE
      postAction:
        type: ARCHIVE
        destinationPrefix: gs://my-bucket/archive/
      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"
          }
    outputs:
      - sink

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

Local filesystem batch with stability probe​

Reads log files from /var/logs/app/, waiting 10 seconds after first seeing each file to ensure it has finished being written.

dag:
  - id: source
    commandName: fssource
    config:
      backend: file
      root: /var/logs/app/
      fileNameRegex: ".*\\.log"
      mode: BOUNDED
      emission:
        type: LINE
      stability:
        enabled: true
        probeDelayMs: 10000
    outputs:
      - sink

  - id: sink
    commandName: stdout
    config:
      encodingType: JSON_OBJECT
  • gcssink: Write records to a Google Cloud Storage bucket
  • s3sink: Write records to Amazon S3
  • kafkasource: Stream-based source for Kafka topics