Decoding and Encoding

Sources read bytes with a decoding configuration and sinks write bytes with an encoding configuration. Both have the same shape: a format and an optional compression.

Introduction

ZephFlow processes data from various sources and destinations, often requiring conversion between byte streams and structured records. A source that reads bytes (Kafka, S3, SQS, files, standard input, and so on) says how to turn them into records with decoding. A sink that writes bytes (Kafka, S3, Kinesis, standard output, and so on) says how to turn records into bytes with encoding.

decoding:            # on a source
  compression: gzip  # optional
  format:
    type: json
    mode: lines

encoding:            # on a sink
  format:
    type: json
    mode: object

Nodes that talk to a structured API, such as JDBC, Elasticsearch or Splunk search, build records directly and take neither.

Source Decoding

Every source that reads bytes requires decoding. A config without it, or with a bad one, fails validation.

• compression is one codec or a list applied in order: gzip, zstd, bzip2, xz, lz4 (frame format), snappy (framing format), deflate (zlib), or auto, which gunzips input that starts with the gzip magic bytes and passes any other through. auto cannot be combined with another codec. With no compression, nothing is decompressed.
• format.type is one of json, text, csv, xml, yaml, avro, protobuf, otlp, parquet, auto. Each takes its own parameters, such as mode for json and framing for text. See Formats for every format's parameters and the records it gives.

Each source page lists what its decoding accepts and how it handles input that fails to decode, for example kafkasource and fssource.

Sink Encoding

encoding takes a format and an optional compression, with these rules:

• format.type is one of json (with mode object, array or lines), csv, kv, text or parquet, as the sink allows. The formats that can only be read (auto, avro, otlp, protobuf, xml, yaml) are rejected.
• compression is absent or one codec. Only gzip can be written, and only by a sink that allows it; auto is rejected, since it describes how a reader guesses.
• A sink that writes one record per message cannot write parquet. A sink that writes many records per file cannot write json mode object or text framing whole, since each holds one record.
• text writes each record's __raw__ string, and a record without a string __raw__ fails. Text and CSV are written as UTF-8.

What each sink can write:

Sink	Unit written	Formats	json modes	compression	Without encoding
kafkasink	One message per record	json, csv, kv	object, array, lines	None	Fails validation
kinesissink	One Kinesis record per record	json, csv	object, array, lines	None	Fails validation
azureeventhubsink	One event per record	json, csv	object, array, lines	None	Fails validation
stdout	One output per record	json, csv	object, array, lines	None	Fails validation
s3sink	One file per batch	json, csv, kv, text, parquet	array, lines	gzip	Fails validation
splunkhecsink	One HEC event per record	json, csv (with header: false and columns), kv, text	object	None	Sends each record as a JSON object event
sqssink	One message per record	json	object	None	Sends each record as a JSON object
pubsubsink	One message per record	json	object	None	Publishes each record as a JSON object

When a sink writes one record per message, json mode array writes a one-element array, lines writes a single line, and csv writes a header line and a value line unless header is false and columns is set.

The format parameters that apply to writing:

format.type	Parameters
json	mode (required)
csv	delimiter (default ,), quote (default "), escape, header (default true), columns (which fields to write, in order; required when header is false). skipLines, commentMarker, a charset other than UTF-8, and quote: "" are rejected
kv	fields (default: all, sorted), pairDelimiter (default a space), cleanFields (replace every key character other than letters, digits and _ with _)
text	framing (required), as when reading
parquet	schema (required): an Avro record schema, as a JSON object or a string holding one. Cannot be combined with compression

Usage Examples

Kafka Source and Sink

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  # Read JSON objects from a Kafka topic, gunzipping the messages that are gzipped
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "input-topic"
      groupId: "my-consumer-group"
      decoding:
        compression: auto
        format:
          type: json
          mode: object
    outputs:
      - "kafka_sink"

  # Write processed records as JSON objects to another Kafka topic
  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka:9092"
      topic: "output-topic"
      encoding:
        format:
          type: json
          mode: object
    outputs: [ ]

S3 Sink Example

When writing data to S3, encoding sets the format of the stored files:

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "input-topic"
      groupId: "my-consumer-group"
      decoding:
        format:
          type: json
          mode: object
    outputs:
      - "success_filter"

  - id: "success_filter"
    commandName: "filter"
    config:
      expression: $.status == 'success'  # only process successful records
    outputs:
      - "s3_sink"

  # Store records as gzipped CSV files in S3
  - id: "s3_sink"
    commandName: "s3sink"
    config:
      regionStr: "us-west-2"
      bucketName: "my-bucket"
      keyName: "data/output/"
      encoding:
        compression: gzip
        format:
          type: csv
    outputs: [ ]

StdIn/StdOut Example

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  # Read each line from standard input as a text record
  - id: "stdin_source"
    commandName: "stdin"
    config:
      decoding:
        format:
          type: text
          framing:
            type: line
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: dict(text=$.__raw__, upper_text=upper($.__raw__))  # text records only have __raw__
    outputs:
      - "stdout_sink"

  - id: "stdout_sink"
    commandName: "stdout"
    config:
      encoding:
        format:
          type: json
          mode: lines
    outputs: [ ]

Best Practices

1. Match the format to your data: Always ensure the decoding or encoding you configure matches the actual data you're reading from or writing to.

2. Set compression when the input is compressed: A source decompresses only what compression says. Use auto when only some input is gzipped.

3. Pay attention to serialization/deserialization:

   • Kafka nodes read and write raw bytes and leave the format to decoding and encoding.
   • Use byte array serializers in the Kafka producers that write to topics ZephFlow reads, for example:

   Properties producerProps = new Properties();
   producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
   producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());