Understanding Data Flows in ZephFlow

ZephFlow uses Directed Acyclic Graphs to define data processing logic, with support for branching, merging, and complex multi-step transformations from source to sink.

What is a DAG?

A Directed Acyclic Graph (DAG) is a fundamental concept in ZephFlow that defines how your data processing pipeline works. Let's break down what this means:

• Directed: Data flows in a specific direction, from one processing step to the next
• Acyclic: The data never flows in circles or loops
• Graph: A network of connected processing steps

Think of a DAG as a flowchart for your data, where each step performs a specific operation, and arrows show where the data goes next.

Data Processing with DAGs

In ZephFlow, a DAG defines the entire processing logic of your application. When you execute a ZephFlow pipeline:

1. Data enters at source nodes (like reading from Kafka or a file)
2. Flows through various transformation nodes (filtering, mapping, etc.)
3. Exits through sink nodes (like writing to a database or file)

The Water Stream Analogy

A helpful way to visualize how data flows through a ZephFlow DAG is to imagine a stream of water:

• Source nodes are like springs where water (data) originates
• Processing nodes are like water filters or treatment plants that change the water
• Connections are like pipes that direct the water flow
• Branching points split the water into multiple streams
• Merging points combine streams back together
• Sink nodes are like reservoirs where the water ultimately collects

Each drop of water (record) follows the defined paths, being transformed along the way until it reaches its destination.

Core Components of a ZephFlow DAG

A ZephFlow DAG consists of three main elements:

1. Nodes: Individual processing steps (commands) in the pipeline
2. Connections: Directed edges that define how data flows between nodes
3. Job Context: Optional configuration that applies to the entire pipeline

Job Context

The jobContext section provides optional global configuration for the entire pipeline:

• otherProperties: Custom key-value properties accessible throughout the pipeline
• metricTags: Tags attached to metrics for monitoring (e.g., env, service)
• logLevel: Sets ZephFlow's log level. Supported values: DEBUG, INFO, WARN, ERROR. Defaults to INFO.
• dlqConfig: Dead letter queue configuration (see Error Handling)

Example with log level:

jobContext:
  logLevel: debug
  metricTags:
    env: "staging"
    service: "data-processor"

Node Structure

Each node in a DAG has these key properties:

• id: A unique identifier for the node
• commandName: The type of operation to perform (e.g., "filter", "eval", "kafkasource")
• config: A configuration map specific to the node (structure varies by command type)
• outputs: A list of node IDs that this node connects to (where data flows next)

DAG Definition Formats

ZephFlow supports two formats for defining DAGs:

YAML Format (for CLI usage)

jobContext:
  # Optional global configuration
  otherProperties:
  metricTags:
  logLevel:
  dlqConfig:

dag:
  - id: "inputNode"
    commandName: "kafkasource"
    config:
      broker: "localhost:9092"
      topic: "input-topic"
      groupId: "my-consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "filterNode"

  - id: "filterNode"
    commandName: "filter"
    config:
      expression: $.status == "success"
    outputs:
      - "transformNode"

  - id: "transformNode"
    commandName: "eval"
    config:
      expression: |
        dict(
          user_id = $.user.id,
          timestamp = $.created_at,
          value = $.amount * 100
        )
    outputs:
      - "outputNode"

  - id: "outputNode"
    commandName: "kafkasink"
    config:
      broker: "localhost:9092"
      topic: "output-topic"
      encodingType: "JSON_OBJECT"
    outputs: [ ]

JSON Format

{
  "jobContext": {
    "otherProperties": {},
    "metricTags": {}
  },
  "dag": [
    {
      "id": "inputNode",
      "commandName": "kafkasource",
      "config": {
        "broker": "localhost:9092",
        "topic": "input-topic",
        "groupId": "my-consumer-group",
        "encodingType": "JSON_OBJECT"
      },
      "outputs": [
        "filterNode"
      ]
    },
    {
      "id": "filterNode",
      "commandName": "filter",
      "config": {
        "expression": "$.status == \"success\""
      },
      "outputs": [
        "transformNode"
      ]
    },
    {
      "id": "transformNode",
      "commandName": "eval",
      "config": {
        "expression": "dict(\n  user_id = $.user.id,\n  timestamp = $.created_at,\n  value = $.amount * 100\n)"
      },
      "outputs": [
        "outputNode"
      ]
    },
    {
      "id": "outputNode",
      "commandName": "kafkasink",
      "config": {
        "broker": "localhost:9092",
        "topic": "output-topic",
        "encodingType": "JSON_OBJECT"
      },
      "outputs": []
    }
  ]
}

Creating a ZephFlow DAG

Let's walk through creating a simple DAG that processes JSON data:

Step 1: Define Your Processing Requirements

For this example, we'll create a DAG that:

1. Reads JSON data from standard input
2. Filters records where the "value" field is greater than 10
3. Transforms the data by adding new fields
4. Outputs the results to standard output

Step 2: Write the DAG Definition

Create a file named simple_data_processor.yml:

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "stdin_source"
    commandName: "stdin"
    config:
      encodingType: "JSON_ARRAY"
    outputs:
      - "filter_node"

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

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

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

Step 3: Run the DAG

Run the DAG with the CLI starter and pipe JSON data to it:

echo '[{"value": 5, "timestamp": 1635724800000}, {"value": 15, "timestamp": 1635724800000}]' | docker run -i \
  -v $(pwd)/simple_data_processor.yml:/app/simple_data_processor.yml \
  fleak/zephflow-clistarter:latest -f /app/simple_data_processor.yml

Expected output (one compact line per record, printed between the log lines):

use an empty line to quit
{"doubled_value":30,"original_value":15,"status":"processed","timestamp":"2021-11-01T00:00:00"}

Creating More Complex DAGs

For more complex processing needs, you can create branching and merging flows. A node that lists several nodes in outputs branches the flow, and several nodes that list the same node in outputs merge into it:

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "stdin_source"
    commandName: "stdin"
    config:
      encodingType: "JSON_ARRAY"
    outputs:
      - "high_value_filter"
      - "standard_value_filter"

  # Branch 1: High-value transactions
  - id: "high_value_filter"
    commandName: "filter"
    config:
      expression: $.value >= 10
    outputs:
      - "high_value_tag"

  - id: "high_value_tag"
    commandName: "eval"
    config:
      expression: dict_merge($, dict(category='high-value', priority='urgent'))
    outputs:
      - "add_processed_at"

  # Branch 2: Standard transactions
  - id: "standard_value_filter"
    commandName: "filter"
    config:
      expression: $.value < 10
    outputs:
      - "standard_value_tag"

  - id: "standard_value_tag"
    commandName: "eval"
    config:
      expression: dict_merge($, dict(category='standard', priority='normal'))
    outputs:
      - "add_processed_at"

  # Both branches merge into the final processing step
  - id: "add_processed_at"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(processed_at=epoch_to_ts_str($.timestamp, "yyyy-MM-dd'T'HH:mm:ss")))
    outputs:
      - "stdout_sink"

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

Save it as complex_data_processor.yml, then run it with the CLI starter and pipe JSON data to it:

echo '[{"value": 5, "timestamp": 1635724800000}, {"value": 15, "timestamp": 1635724800000}]' | docker run -i \
  -v $(pwd)/complex_data_processor.yml:/app/complex_data_processor.yml \
  fleak/zephflow-clistarter:latest -f /app/complex_data_processor.yml

Expected output:

{"processed_at":"2021-11-01T00:00:00","category":"high-value","priority":"urgent","value":15,"timestamp":1635724800000}
{"processed_at":"2021-11-01T00:00:00","category":"standard","priority":"normal","value":5,"timestamp":1635724800000}

ZephFlow DAG Rules

ZephFlow enforces several rules to ensure your DAGs are valid and can process data correctly:

1. Entry Points depend on how the DAG is executed:

   • For standalone processing: A DAG must have exactly one source node.
   • In the Workflow Builder, sample runs can start from Test Data connected to the processing graph. This canvas input is separate from the persisted processing nodes, so the absence of a source command does not mean the builder has no input or require adding a source node. Use the available test-data and connection context; a processing-node list alone does not show every builder input connection. Production source configuration is a separate deployment step; see Workflow Lifecycle.

2. Source Nodes cannot have incoming connections

3. Sink Nodes must be terminal nodes (cannot have outgoing connections)

4. Node IDs:

   • Each node must have a unique ID within the DAG

5. Acyclic Requirement:

   • The graph cannot contain cycles (data can't loop back to nodes it already passed through)

6. Non-empty DAG:

   • A DAG must contain at least one node

Read Next

As you get more comfortable with ZephFlow, you can explore:

• More advanced node types and transformations
• Error handling with assertion nodes and dead-letter queues
• Integration with various data sources and sinks
• Monitoring and observability features

Further Reading

• Fleak Eval Expression Language (FEEL) Guide
• Node Types Reference
• SQL Transformations in ZephFlow