bask 0.2.1

Build Tasks: an async task-queue pipeline engine with pluggable IO, columnar formats, and predefined tasks
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

/*
 * SPDX-FileCopyrightText: 2026 Wavelens GmbH <info@wavelens.io>
 *
 * SPDX-License-Identifier: MIT OR Apache-2.0
 */

//! Predefined tasks from `bask::tasks`: a large record batch is split into fixed-row
//! `Piece`s by a `Chunker`, then a `RowBatch` router re-aggregates them into groups of at
//! least 100 rows, flushing the trailing group. The pipeline only wires the stages.
use std::sync::Arc;

use arrow::array::{ArrayRef, Int64Array};
use arrow::record_batch::RecordBatch;
use bask::prelude::*;
use bask::tasks::{Batch, Chunker, Piece, RowBatch, Whole};

fn batch(rows: i64) -> RecordBatch {
    let col = Arc::new(Int64Array::from((0..rows).collect::<Vec<_>>())) as ArrayRef;
    RecordBatch::try_from_iter([("n", col)]).unwrap()
}

/// Routes each 32-row piece into the row-count aggregator.
struct Feed;
#[async_trait]
impl Worker for Feed {
    type Task = Piece;
    async fn process(&self, piece: &Piece, ctx: &Context) -> anyhow::Result<()> {
        ctx.route::<RowBatch<100>>(piece.0.clone()).await?;
        Ok(())
    }
}

/// Consumes each re-aggregated group.
struct Show;
#[async_trait]
impl Worker for Show {
    type Task = Batch;
    async fn process(&self, group: &Batch, _ctx: &Context) -> anyhow::Result<()> {
        println!("group of {} rows", group.0.num_rows());
        Ok(())
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let report = Engine::builder()
        .worker(Chunker::<32>)
        .worker(Feed)
        .worker(Show)
        .router::<RowBatch<100>>()
        .concurrency(1)
        .seed(Whole(batch(250)))
        .run()
        .await?;

    println!(
        "groups emitted: {}",
        report.output::<RowBatch<100>>().unwrap()
    );
    Ok(())
}