DEV Community: chenquan

ArkFlow v0.3.0 is now available!

chenquan — Sat, 10 May 2025 04:34:05 +0000

ArkFlow is a high-performance Rust stream processing engine that provides powerful data stream processing capabilities, supporting various input/output sources and processors.

We are thrilled to announce a significant update to ArkFlow! ArkFlow is a high-performance stream processing engine built on Rust, dedicated to providing powerful and easy-to-use data stream processing capabilities. The latest version brings many exciting new features and significant enhancements designed to help you process complex data streams more efficiently.

In this blog post, we will delve into these new features, from broader data source connectivity and more powerful processing capabilities with custom extensions, to landmark distributed computing experimental support, and continuous performance and usability optimizations.

more:https://arkflow-rs.com/blog/2025/05/09/v0.3.0-release

ArkFlow - High-performance Rust stream processing engine

chenquan — Wed, 12 Mar 2025 07:02:23 +0000

ArkFlow

github: https://github.com/chenquan/arkflow

High-performance Rust stream processing engine, providing powerful data stream processing capabilities, supporting multiple input/output sources and processors.

Features

High Performance: Built on Rust and Tokio async runtime, offering excellent performance and low latency
Multiple Data Sources: Support for Kafka, MQTT, HTTP, files, and other input/output sources
Powerful Processing Capabilities: Built-in SQL queries, JSON processing, Protobuf encoding/decoding, batch processing, and other processors
Extensible: Modular design, easy to extend with new input, output, and processor components

Installation

Building from Source

# Clone the repository
git clone https://github.com/chenquan/arkflow.git
cd arkflow

# Build the project
cargo build --release

# Run tests
cargo test

Quick Start

Create a configuration file config.yaml:

logging:
  level: info
streams:
  - input:
      type: "generate"
      context: '{ "timestamp": 1625000000000, "value": 10, "sensor": "temp_1" }'
      interval: 1s
      batch_size: 10

    pipeline:
      thread_num: 4
      processors:
        - type: "json_to_arrow"
        - type: "sql"
          query: "SELECT * FROM flow WHERE value >= 10"
        - type: "arrow_to_json"

    output:
      type: "stdout"

Run ArkFlow:

./target/release/arkflow --config config.yaml

Configuration Guide

ArkFlow uses YAML format configuration files, supporting the following main configuration items:

Top-level Configuration

logging:
  level: info  # Log level: debug, info, warn, error

streams:       # Stream definition list
  - input:      # Input configuration
      # ...
    pipeline:   # Processing pipeline configuration
      # ...
    output:     # Output configuration
      # ...

Input Components

ArkFlow supports multiple input sources:

Kafka: Read data from Kafka topics
MQTT: Subscribe to messages from MQTT topics
HTTP: Receive data via HTTP
File: Read data from files
Generator: Generate test data
SQL: Query data from databases

Example:

input:
  type: kafka
  brokers:
    - localhost:9092
  topics:
    - test-topic
  consumer_group: test-group
  client_id: arkflow
  start_from_latest: true

Processors

ArkFlow provides multiple data processors:

JSON: JSON data processing and transformation
SQL: Process data using SQL queries
Protobuf: Protobuf encoding/decoding
Batch Processing: Process messages in batches

Example:

pipeline:
  thread_num: 4
  processors:
    - type: json_to_arrow
    - type: sql
      query: "SELECT * FROM flow WHERE value >= 10"
    - type: arrow_to_json

Output Components

ArkFlow supports multiple output targets:

Kafka: Write data to Kafka topics
MQTT: Publish messages to MQTT topics
HTTP: Send data via HTTP
File: Write data to files
Standard Output: Output data to the console

Example:

output:
  type: kafka
  brokers:
    - localhost:9092
  topic: output-topic
  client_id: arkflow-producer

Examples

Kafka to Kafka Data Processing

streams:
  - input:
      type: kafka
      brokers:
        - localhost:9092
      topics:
        - test-topic
      consumer_group: test-group

    pipeline:
      thread_num: 4
      processors:
        - type: json_to_arrow
        - type: sql
          query: "SELECT * FROM flow WHERE value > 100"
        - type: arrow_to_json

    output:
      type: kafka
      brokers:
        - localhost:9092
      topic: processed-topic

Generate Test Data and Process

streams:
  - input:
      type: "generate"
      context: '{ "timestamp": 1625000000000, "value": 10, "sensor": "temp_1" }'
      interval: 1ms
      batch_size: 10000

    pipeline:
      thread_num: 4
      processors:
        - type: "json_to_arrow"
        - type: "sql"
          query: "SELECT count(*) FROM flow WHERE value >= 10 group by sensor"
        - type: "arrow_to_json"

    output:
      type: "stdout"

License

ArkFlow is licensed under the Apache License 2.0.

sqltrace: A low-code intrusion library that provides SQL tracing capabilities

chenquan — Thu, 15 Sep 2022 13:29:12 +0000

sqltrace

A low-code intrusion library that provides SQL tracing capabilities, suitable for any relational database (Sqlite3, MySQL, Oracle, SQL Server, PostgreSQL, TiDB, etc.) and ORM libraries for various relational database (gorm, xorm, sqlx, etc.)

chenquan / sqltrace

A low-code intrusion library that provides SQL tracing capabilities, suitable for any relational database (Sqlite3, MySQL, Oracle, SQL Server, PostgreSQL, TiDB, etc.) and ORM libraries for various relational database (gorm, xorm, sqlx, etc.)

sqltrace

A low-code intrusion library that provides SQL tracing capabilities, suitable for any relational database (Sqlite3, MySQL, Oracle, SQL Server PostgreSQL, TiDB, etc.) and ORM libraries for various relational database (gorm, xorm, sqlx, etc.)

😜installation

go get -u github.com/chenquan/sqltrace

👏how to use

native

package main
import (
    "context"
    "database/sql"
    "fmt"

    "github.com/chenquan/sqltrace"
    "github.com/mattn/go-sqlite3"
    _ "github.com/mattn/go-sqlite3"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

func main() {
    // Create a sqlite3 driver with link tracking
    driver := sqltrace.NewDriver(sqltrace.Config{
        Name:           "sqlite3_trace",
        DataSourceName: "sqlite3",
        Endpoint:       "http://localhost:14268/api/traces",
        Sampler:        1,
        Batcher:        "jaeger",
    }, &sqlite3.SQLiteDriver{})
    defer sqltrace.StopAgent()

    // register new driver
    sql.Register("sqlite3_trace", driver)

    // open database
    db, err := sql.Open("sqlite3_trace", "identifier.sqlite")
    if err != nil {
        panic(err)
    }

    tracer := otel.

…

View on GitHub

😜installation

go get -u github.com/chenquan/sqltrace

👏how to use

native

package main

import (
    "context"
    "database/sql"
    "fmt"

    "github.com/chenquan/sqltrace"
    "github.com/mattn/go-sqlite3"
    _ "github.com/mattn/go-sqlite3"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

func main() {
    // Create a sqlite3 driver with link tracking
    driver := sqltrace.NewDriver(sqltrace.Config{
        Name:           "sqlite3_trace",
        DataSourceName: "sqlite3",
        Endpoint:       "http://localhost:14268/api/traces",
        Sampler:        1,
        Batcher:        "jaeger",
    }, &sqlite3.SQLiteDriver{})
    defer sqltrace.StopAgent()

    // register new driver
    sql.Register("sqlite3_trace", driver)

    // open database
    db, err := sql.Open("sqlite3_trace", "identifier.sqlite")
    if err != nil {
        panic(err)
    }

    tracer := otel.GetTracerProvider().Tracer("sqlite3_trace")
    ctx, span := tracer.Start(context.Background(),
        "test",
        trace.WithSpanKind(trace.SpanKindClient),
    )
    defer span.End()

    db.ExecContext(ctx, `CREATE TABLE t
(
    age  integer,
    name TEXT
)`)
    db.ExecContext(ctx, "insert into t values (?,?)", 1, "chenquan")

    // transaction
    tx, err := db.BeginTx(ctx, nil)
    stmt, err := tx.PrepareContext(ctx, "select age+1 as age,name from t where age = ?;")
    stmt.QueryContext(ctx, 1)
    tx.Commit()

    rows, err := db.QueryContext(ctx, "select  age+1 as age,name from t;")
    for rows.Next() {
        var age int
        var name string
        err := rows.Scan(&age, &name)
        if err != nil {
            fmt.Println(err)
        }

        fmt.Println(age, name)
    }
}

gorm

package main

import (
    "context"
    "database/sql"
    "fmt"

    "github.com/chenquan/sqltrace"
    "github.com/mattn/go-sqlite3"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
    "gorm.io/driver/sqlite"
    "gorm.io/gorm"
)

func main() {
    // Create a sqlite3 driver with link tracking
    driver := sqltrace.NewDriver(sqltrace.Config{
        Name:           "sqlite3_trace",
        DataSourceName: "sqlite3",
        Endpoint:       "http://localhost:14268/api/traces",
        Sampler:        1,
        Batcher:        "jaeger",
    }, &sqlite3.SQLiteDriver{})
    defer sqltrace.StopAgent()

    // register new driver
    sql.Register("sqlite3_trace", driver)

    db, err := gorm.Open(sqlite.Dialector{
        DriverName: "sqlite3_trace",
        DSN:        "identifier.sqlite",
    }, &gorm.Config{})
    if err != nil {
        panic(err)
    }
    tracer := otel.GetTracerProvider().Tracer("sqlite3_trace")
    ctx, span := tracer.Start(context.Background(),
        "test",
        trace.WithSpanKind(trace.SpanKindClient),
    )
    defer span.End()

    type T struct {
        Age  int    `db:"age"`
        Name string `db:"name"`
    }
    db = db.WithContext(ctx)
    db.Migrator().AutoMigrate(&T{})
    db.Create(&T{
        Age:  1,
        Name: "chenquan",
    })

    rows, err := db.Raw("select  age+1 as age,name from ts;").Rows()
    if err != nil {
        panic(err)
    }

    for rows.Next() {
        var age int
        var name string
        err := rows.Scan(&age, &name)
        if err != nil {
            fmt.Println(err)
        }

        fmt.Println(age, name)
    }
    db.Commit()
}

⭐star

If you like or are using this project to learn or start your solution, please give it a star⭐. Thanks!

A tool for showing disk usage. (Linux, macOS and Windows)

chenquan — Thu, 11 Aug 2022 00:04:00 +0000

Use the go to develop a tool to display the current disk usage.
https://github.com/chenquan/diskusage
If you like or are using this project to learn or start your solution, please give it a star⭐. Thanks!

❗installation

go install github.com/chenquan/diskusage@latest

👏how to use

$ diskusage -h
A tool for showing disk usage.

Usage:
  diskusage [flags]

Flags:
  -a, --all             display all directories, otherwise only display folders whose usage size is not 0
  -c, --color string    set color output mode. optional: auto, always, ignore (default "auto")
  -d, --depth int       shows the depth of the tree directory structure (default 1)
      --dir string      dir path (default "./")
  -f, --filter string   regular expression filter (default ".+")
  -h, --help            help for diskusage
  -t, --type strings    only count certain types of files  (default all)
  -u, --unit string     displayed units. optional: B(Bytes), K(KB), M(MB), G(GB), T(TB) (default "M")

👀example

Only files named Doc or docx are counted: diskusage -t doc,docx or diskusage -f ".+\.(doc|docx)$"
The maximum display unit is GM: diskusage -u G
Supports color output to pipeline: diskusage -c always | less -R or diskusage -c always | more