Browse documentation

#Aggregations

You can aggregate each group of the groupBy function with one of the aggregation functions.

#Aggregation functions

• COUNT()
• AVERAGE()
• SUM()
• MIN()
• MAX()
• COLLECT()
• COLLECT_UNIQUE()
• FIRST()
• LAST()
• STRING_AGG()

All aggregation functions implement AggregatingFunction. aggregate(Rows $rows, array $indices, Backend $backend) folds the rows of one group into the state. merge(array $partial, Backend $backend) folds in the partial() of another aggregator of the same function. Group by merges after a spill.

#Custom aggregators

A custom aggregator also describes itself, because a worker builds it again from describe(). It gives its state as a partial: the plain values that a worker returns, so that the parent or a later task can merge them:

<?php

final class LongestString implements AggregatingFunction
{
    private ?string $longest = null;

    public function __construct(private readonly Reference $ref) {}

    public static function fromDescription(array $data, Rebuild $rebuild): static
    {
        return new self($rebuild->of($data['ref'], Reference::class));
    }

    public function describe(): ?Description
    {
        return Description::compose(self::class, ['ref' => $this->ref]);
    }

    public function aggregate(Rows $rows, array $indices, Backend $backend): void
    {
        $column = $rows->column($this->ref->to());

        foreach ($indices as $index) {
            $value = $column->value($index);

            if (is_string($value) && strlen($value) > strlen($this->longest ?? '')) {
                $this->longest = $value;
            }
        }
    }

    public function partialType(): StructureType
    {
        return type_structure(['longest' => type_optional(type_string())]);
    }

    public function partial(): array
    {
        return ['longest' => $this->longest];
    }

    public function merge(array $partial, Backend $backend): void
    {
        $longest = $this->partialType()->assert($partial)['longest'];

        if ($longest !== null && strlen($longest) > strlen($this->longest ?? '')) {
            $this->longest = $longest;
        }
    }

    public function withChildren(array $children): static
    {
        return new self($children[0]);   // a fresh instance: no state crosses
    }

    // children(), resolved(), deterministic(), references(), returns(), outputName(), value() as before
}

partialType() is valid after the aggregator is resolved. partial() of a new instance is the initial state, and a merge of it is a no-op. Partials of two tasks merge in task order. Order-sensitive aggregators (first(), last(), collect()) depend on this order. A describe() that returns null keeps the group by in the parent process.

#Float arithmetic

sum() and average() add floats as IEEE 754 doubles with compensated summation. exact: true changes to decimal arithmetic:

<?php

use function Flow\ETL\DSL\{df, from_array, ref, sum};

df()->read(from_array([['a' => 0.1], ['a' => 0.2]]))->aggregate([sum(ref('a'))])->fetch();              // 0.30000000000000004
df()->read(from_array([['a' => 0.1], ['a' => 0.2]]))->aggregate([sum(ref('a'), exact: true)])->fetch(); // 0.3

#Example

<?php

use Flow\ETL\Foundation\Calculator\Rounding;

use function Flow\ETL\Adapter\CSV\from_csv;
use function Flow\ETL\DSL\{average, concat, data_frame, lit, max, min, ref, sum, to_output};

data_frame()
    ->read(from_csv(__DIR__ . '/power-plant-daily.csv', separator: ';'))
    ->withColumn('production_kwh', ref('Produkcja(kWh)'))
    ->withColumn('consumption_kwh', ref('Zużycie(kWh)'))
    ->withColumn('date', ref('Zaktualizowany czas')->toDate('Y/m/d')->dateFormat('Y/m'))
    ->select('date', 'production_kwh', 'consumption_kwh')
    ->groupBy([ref('date')])
    ->aggregate(
        average(ref('production_kwh')),
        average(ref('consumption_kwh')),
        min(ref('production_kwh')),
        min(ref('consumption_kwh')),
        max(ref('production_kwh')),
        max(ref('consumption_kwh')),
        sum(ref('production_kwh')),
        sum(ref('consumption_kwh'))
    )
    ->withColumn('production_kwh_avg', ref('production_kwh_avg')->round(lit(2)))
    ->withColumn('consumption_kwh_avg', ref('consumption_kwh_avg')->round(lit(2)))
    ->withColumn('production_kwh_min', ref('production_kwh_min')->round(lit(2)))
    ->withColumn('consumption_kwh_min', ref('consumption_kwh_min')->round(lit(2)))
    ->withColumn('production_kwh_max', ref('production_kwh_max')->round(lit(2)))
    ->withColumn('consumption_kwh_max', ref('consumption_kwh_max')->round(lit(2)))
    ->withColumn('production_kwh_sum', ref('production_kwh_sum')->round(lit(2)))
    ->withColumn('consumption_kwh_sum', ref('consumption_kwh_sum')->round(lit(2)))
    ->withColumn('consumption', ref('consumption_kwh_sum')->divide(ref('production_kwh_sum'), 4, Rounding::HALF_UP))
    ->withColumn('consumption', ref('consumption')->multiply(lit(100))->round(lit(2)))
    ->withColumn('consumption', concat(ref('consumption'), lit('%')))
    ->write(to_output(truncate: false))
    ->run();

Found a typo or an outdated section? Edit this page on GitHub ↗