Skip to main content

Quick Start Guide

Get started with ZephFlow by running it as a standalone Docker container with the CLI starter.

ZephFlow is a flexible data processing framework that allows you to build and execute powerful data transformation pipelines. This guide will help you get started with ZephFlow and explore its basic functionality.

Introduction​

ZephFlow enables you to create data processing workflows using a directed acyclic graph (DAG) structure. You can define data flows that filter, transform, and process data using the Fleak Eval Expression Language (FEEL), a flexible expression language designed for data manipulation.

With ZephFlow, you can:

  • Create data processing pipelines with filtering, transformation, and validation
  • Run pipelines as a standalone process
  • Process data in various formats
  • Apply complex transformations using a powerful expression language
info

To understand what is a DAG and how ZephFlow uses DAG to express processing logic, check the DAG structure docs.

Ways to Use ZephFlow​

ZephFlow runs as a standalone data pipeline process using the CLI starter.

Running ZephFlow as a Standalone Process with CLI Starter​

The CLI starter provides a convenient way to run ZephFlow pipelines in a Docker container. This section guides you through creating and running a simple data processing pipeline using the CLI starter.

Overview​

The ZephFlow CLI starter:

  • Reads a DAG definition file that defines your data pipeline
  • Processes data according to the defined pipeline
  • Supports various input/output modes defined in your pipeline definition (stdin/stdout, files, Kafka, etc.)

Installation​

Pull the CLI starter docker image:

docker pull fleak/zephflow-clistarter:latest

Quick Start Example​

Let's create a simple pipeline that reads JSON data from standard input, filters records, transforms them, and outputs the results to standard output.

Step 1: Create a sample DAG definition file​

Create a file named simple_pipeline.yml with the following content:

jobContext:
  otherProperties: { }
  metricTags:
    env: "default_env"
    service: "default_service"
  dlqConfig: null

dag:
  - id: "stdin_source"
    commandName: "stdin"
    config:
      decoding:
        format:
          type: json
          mode: array
    outputs:
      - "filter_node"

  - id: "filter_node"
    commandName: "filter"
    config:
      expression: $.value > 10
    outputs:
      - "transform_node"

  - id: "transform_node"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(
          processed=true,
          timestamp_iso=epoch_to_ts_str($.timestamp, "yyyy-MM-dd'T'HH:mm:ss"),
          doubled_value=$.value * 2
        ))
    outputs:
      - "stdout_sink"

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

This DAG reads a collection of JSON objects from your standard input, keeps only the ones with values greater than 10, and enhances them with useful additions. Each record gets a "processed" flag, a human-readable timestamp converted from Unix time, and a "doubled_value" field that's twice the original. The processed output records are then printed out in your terminal.

Step 2: Run the ZephFlow CLI starter​

Run the Docker container, mapping your DAG definition file and allowing stdin/stdout to flow through:

docker run -i \
  -v $(pwd)/simple_pipeline.yml:/app/simple_pipeline.yml \
  fleak/zephflow-clistarter:latest -f /app/simple_pipeline.yml

The -i flag is critical as it keeps stdin open, allowing you to send data to the container. When the container starts, the stdin source prints use an empty line to quit.

Step 3: Send test data and observe results​

Once the container is running, you can send JSON data via stdin. In your terminal, paste the following JSON as a single line and press Enter. The stdin source parses each input line separately, so the whole array must be on one line:

[{"value": 5, "timestamp": 1635724800000}, {"value": 15, "timestamp": 1635724800000}, {"value": 25, "timestamp": 1635724800000}]

After entering the data, press Ctrl+D (or enter an empty line) to signal the end of input.

Between the log lines, you should see output similar to:

{"doubled_value":30,"processed":true,"timestamp_iso":"2021-11-01T00:00:00","value":15,"timestamp":1635724800000}
{"doubled_value":50,"processed":true,"timestamp_iso":"2021-11-01T00:00:00","value":25,"timestamp":1635724800000}

Notice that only records with value > 10 appear in the output, and they include the additional fields defined in the transform node.

Data Transformation in ZephFlow​

ZephFlow provides multiple approaches for data transformation, filtering, and validation. The framework is built with a modular command architecture, where each command in the DAG is a standalone unit that performs a specific function.

Transformation Approaches​

Fleak Eval Expression Language (FEEL)​

FEEL provides a concise, expressive way to manipulate data structures. Common FEEL-based commands include:

  • filter: Selectively process records based on conditions

    $.status == "success" and $.response_time < 500

  • eval: Transform data structures

    dict(
      user_id=$.user.id,
      total_amount=$.items[0].price * $.items[0].quantity
    )

  • assertion: Validate data against conditions

    $.user_id != null and $.transaction_id != null

Source and Sink Commands​

ZephFlow pipelines typically begin with a source command and end with a sink command:

  • Source commands bring data into the pipeline (e.g., file, stdin, Kafka)
  • Sink commands output the processed data (e.g., stdout, Kafka, database)

See the ZephFlow nodes index for the available sources and sinks.

Example Pipeline Using Multiple Commands​

Here's an example that demonstrates how various commands can be combined in a pipeline:

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "kafka_source" # Source command
    commandName: "kafkasource"
    config:
      broker: "my_broker"
      topic: "input_topic"
      groupId: "test_group"
      decoding:
        format:
          type: json
          mode: object
    outputs:
      - "error_filter"

  - id: "error_filter" # FEEL filter command
    commandName: "filter"
    config:
      expression: $.level == 'ERROR'
    outputs:
      - "add_timestamp"

  - id: "add_timestamp" # FEEL transform
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(timestamp_ms=ts_str_to_epoch($.timestamp, 'yyyy-MM-dd HH:mm:ss')))
    outputs:
      - "kafka_sink"

  - id: "kafka_sink" # Sink command
    commandName: "kafkasink"
    config:
      broker: "my_broker"
      topic: "output_topic"
      encoding:
        format:
          type: json
          mode: object
    outputs: [ ]

Next Steps​

After getting familiar with the basics of ZephFlow, you might want to: