Skip to main content

Stdin Source Node

The stdin node reads lines from the standard input of the ZephFlow process, parses each line with the configured encoding, and emits the resulting records. It reads until end of input or the first empty line, then ends. Internal ZephFlow engine command for local and CLI runs. Not offered in Fleak Data Workflow and not supported in Fleak Data Workflow workflows; do not add it to a workflow.

Internal engine command

Internal ZephFlow engine command for local and CLI runs. Not offered in Fleak Data Workflow and not supported in Fleak Data Workflow workflows; do not add it to a workflow.

Key Features​

  • Line-by-line parsing: Each input line is parsed on its own with the configured encodingType
  • One batch per run: All lines are read first and emitted together; the source then ends
  • Dead-lettering: A line that fails to parse goes to the DLQ when one is configured

Config Object​

The full config object for the stdin node (StdInSourceDto.Config):

FieldTypeRequiredDefaultDescription
encodingTypeEncodingTypeYes—Format of each input line. One of CSV, JSON_OBJECT, JSON_ARRAY, JSON_OBJECT_LINE, STRING_LINE, TEXT, XML. PARQUET is rejected

Enum values are case-sensitive: json_object fails DAG compilation. See Encoding Types for how each encoding maps input to records.

How Input Is Read​

End of Input​

When the node starts it prints use an empty line to quit to stdout. It then reads lines until standard input reaches end of file (for example Ctrl+D, or the end of a piped file) or until the first empty or whitespace-only line. Lines after that empty line are ignored. All lines read are then emitted as one batch, and the source ends. Nothing is emitted while lines are still being typed.

When running in Docker, pass -i so that the container's standard input stays open.

Per-Line Parsing​

Each line is parsed separately. As a consequence:

  • A JSON_ARRAY or JSON_OBJECT value must be on a single line. In a pretty-printed array spread over several lines, every line fails to parse and no records are emitted.
  • CSV yields no records, because each line is parsed as its own CSV document and a single line is only a header.
  • STRING_LINE and TEXT turn each line into a record with the line in __raw__.

Parse Failures​

A line that fails to parse is logged at ERROR level with its content in base64. When jobContext.dlqConfig is set, the line is written to the DLQ with the stdin node's ID in nodeId; otherwise it is dropped. The other lines are not affected.

Job Context​

The DAG definition must contain a jobContext block, which may be empty (jobContext: {}). Without it, the CLI starter fails with a NullPointerException before the DAG runs.

Examples​

DAG Definition​

Save as stdin_example.yml:

jobContext: {}

dag:
  - id: "stdin_source"
    commandName: "stdin"
    config:
      encodingType: "JSON_OBJECT"
    outputs:
      - "stdout_sink"

  - id: "stdout_sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"
    outputs: []

Run it with the CLI starter and pipe two lines to it:

printf '{"user": "alice", "status": 200}\n{"user": "bob", "status": 500}\n' | docker run -i \
  -v $(pwd)/stdin_example.yml:/app/stdin_example.yml \
  fleak/zephflow-clistarter:latest -f /app/stdin_example.yml

Between the log lines, the output contains:

use an empty line to quit
{"user":"alice","status":200}
{"user":"bob","status":500}
  • stdout: Print records to standard output
  • filesource: Read a single local file
  • fssource: Read files from S3, GCS, Azure Blob Storage, SFTP, or the local filesystem