ArkFlow Documentation

As of version 0.5.0, ArkFlow is a single-binary, configuration-based (YAML), and plugin-oriented stream processing engine built with Rust 1.97 and DataFusion 54.1.

Key Technical Strengths:

• Data Model: Uses Apache Arrow RecordBatch (MessageBatch), providing a columnar data model unlike row-based JSON engines.
• SQL Processing: Powered by DataFusion 54, supporting aggregations, window functions, joins, UDFs, and temporary tables.
• Input Durability: Features Input-level Write-Ahead Logging (WAL) providing at-least-once delivery, ack-gated cursors, crash recovery, and S3 backend support.
• Extensibility: Supports Python UDFs (via PyO3) and VRL (Vector Remap Language) for scripting, alongside a wide range of plugins (14 inputs, 12 outputs, 6 processors, etc.).
• CLI Support: Includes components list/show/schema commands to support IDE auto-completion.

The tumbling_window component is built on the BaseWindow component and uses the following mechanism:

• Grouping: Messages are grouped by input name using an internal RwLock<HashMap<String, Arc<RwLock<VecDeque>>>> structure.
• Triggering: A background timer using the Tokio async runtime triggers window processing at the configured interval.
• Batching: When the interval elapses, messages are batched and concatenated using Arrow's concat_batches for high-performance processing.
• Joins: If configured, SQL joins are performed via DataFusion with parallel decoding.
• Safety: The component implements backpressure handling to prevent memory overflow and uses cancellation tokens for graceful shutdowns.

The Pulsar input component automatically extracts metadata for every message. The following keys are available in the message payload:

• __meta_topic: The source topic.
• __meta_message_id: The unique Pulsar message ID.
• __meta_publish_time: The timestamp when the message was published.
• __meta_ingest_time: The timestamp when the message was ingested by ArkFlow.

The schema_registry codec is used to decode Protobuf messages that follow the Confluent wire format. It resolves the schema ID embedded in the message from a Schema Registry at runtime. It supports schema evolution by fetching and caching each schema version (ID) as it is encountered, allowing multiple versions to coexist in a single stream.

When to use

• When consuming Protobuf messages produced by Confluent serializers (the Kafka ecosystem standard).
• In CDC or pipeline scenarios where source schemas evolve and producers register new versions to a Schema Registry.

Wire Format

The codec expects the following structure: [0x00 magic][4-byte big-endian schema id][Protobuf payload]

The codec validates the magic byte, extracts the ID, resolves it to a schema via the registry, and then decodes the payload.

The backpressure mechanism uses tokio::sync::Notify to transition from periodic polling (sleeping) to an event-driven model. This ensures that when the output stage advances the next_seq, the processor workers are notified immediately rather than waiting for a sleep timer to expire.

Key Logic:

• Trigger: The do_output component calls next_seq_notify.notify_one() immediately after incrementing next_seq via fetch_add(1, Release).
• Waiting Pattern: Processors use a "check-then-await" pattern to minimize latency and avoid unnecessary blocking. They first acquire a notification future, check if the pending count is still above the threshold, and only then .await the notification.
• Lifecycle: The next_seq_notify (an Arc<Notify>) is a field within the Stream struct, initialized during Stream::new and cloned to workers in run_inner.

The memory input component is used to read data from an in-memory message queue. It allows you to seed a workflow with an initial list of messages.

- input:
    type: "memory"
    messages:
      - "Hello"
      - "World"

The SQL input component allows you to query data from various sources using SQL statements. It supports local file-based data sources (like Parquet, CSV, or JSON) and database connections (like MySQL, Postgres, or SQLite). You can also enable distributed computing via Ballista for large-scale queries.

Core configuration keys:

• select_sql: (Required) The SQL query string to execute.
• input_type: (Required) An object specifying the source type and its specific connection or file details.
• ballista: (Optional) Configuration for distributed execution using a Ballista server.

The debezium_json codec does not manage its own offsets or checkpoints. CDC position tracking is handled entirely by the existing Kafka input's ack-gated offset mechanism.

Offsets are advanced via the input-durability component's ack-gated source-commit logic: when downstream writes are confirmed and the Kafka input's ack is triggered, the Kafka offset advances according to standard input-durability semantics.

To enable UPSERT (Insert or Update) functionality, provide the primary_key configuration. This allows for idempotent writes by performing INSERT ... ON CONFLICT/DUPLICATE KEY UPDATE (PostgreSQL/MySQL) or INSERT OR REPLACE (SQLite) based on the specified key(s). You can use a single column or a composite key (array of columns).

# PostgreSQL with UPSERT on a single ID
- output:
    type: "sql"
    type: "postgresql"
    dsn: "postgresql://user:pass@localhost:5432/production"
    table: "metrics"
    primary_key: "id"
    batch_size: 1000

# PostgreSQL with Composite Primary Key
- output:
    type: "sql"
    type: "postgresql"
    dsn: "postgresql://user:pass@postgres:5432/app"
    table: "daily_stats"
    primary_key: ["device_id", "date"]
    batch_size: 200