Master Efficient PHP Data Pipelines with the Low‑Memory Flow Framework

Introduction

Flow is a strongly‑typed PHP library for building large‑scale, low‑memory data‑processing pipelines. It can read from and write to various sources such as databases, CSV, JSON, or native arrays.

Core Features

Strongly‑typed PHP core with minimal memory footprint.

Supports construction of highly scalable pipelines.

Allows data movement between multiple source and target types.

Runs on Linux, macOS and experimental Windows.

Quick Start

Installation

composer require flow-php/etl

Array Example

Read data directly from a PHP array, useful when data is already in memory (e.g., API responses or test fixtures).

<?php
declare(strict_types=1);

use function Flow\ETL\DSL\{data_frame, from_array, to_output};

require __DIR__ . '/vendor/autoload.php';

data_frame()
    ->read(from_array([
        ['id' => 1],
        ['id' => 2],
        ['id' => 3],
        ['id' => 4],
        ['id' => 5],
    ]))
    ->collect()
    ->write(to_output(truncate: false))
    ->run();

CSV File Processing

Read a CSV file, compute daily revenue, and write the aggregated result to a new CSV.

<?php
declare(strict_types=1);

use function Flow\ETL\Adapter\CSV\{from_csv, to_csv};
use function Flow\ETL\DSL\{data_frame, lit, ref, sum, to_output};
use Flow\ETL\Filesystem\SaveMode;

require __DIR__ . '/vendor/autoload.php';

data_frame()
    ->read(from_csv(__DIR__ . '/orders_flow.csv'))
    ->select('created_at', 'total_price', 'discount')
    ->withEntry('created_at', ref('created_at')->cast('date')->dateFormat('Y/m'))
    ->withEntry('revenue', ref('total_price')->minus(ref('discount')))
    ->select('created_at', 'revenue')
    ->groupBy('created_at')
    ->aggregate(sum(ref('revenue')))
    ->sortBy(ref('created_at')->desc())
    ->withEntry('daily_revenue', ref('revenue_sum')->round(lit(2))->numberFormat(lit(2)))
    ->drop('revenue_sum')
    ->write(to_output(truncate: false))
    ->withEntry('created_at', ref('created_at')->toDate('Y/m'))
    ->mode(SaveMode::Overwrite)
    ->write(to_csv(__DIR__ . '/daily_revenue.csv'))
    ->run();

Reference

Project homepage: https://flow-php.com