Flow PHP

DataFrame

FinalYes
Tags
type

Aggregations = list<AggregatingFunction>

GroupByReferences = list<string|Reference>

SortReferences = list<string|Reference>

Methods

__construct()  : mixed
aggregate()  : self
batchBy()  : self
Merge/Split Rows yielded by Extractor into batches but keep those with common value in given column together.
batchSize()  : self
Merge/Split Rows yielded by Extractor into batches of given size.
cache()  : self
Start processing rows up to this moment and put each instance of Rows into previously defined cache.
collect()  : self
Before transforming rows, collect them and merge into single Rows instance.
collectRefs()  : self
This method allows to collect references to all entries used in this pipeline.
constrain()  : self
count()  : int
crossJoin()  : self
display()  : string
drop()  : self
Drop given entries.
dropDuplicates()  : $this
duplicateRow()  : self
fetch()  : Rows
Be aware that fetch is not memory safe and will load all rows into memory.
filter()  : self
filterPartitions()  : self
filters()  : self
forEach()  : void
get()  : Generator<string|int, Rows>
Yields each row as an instance of Rows.
getAsArray()  : Generator<string|int, array<string|int, array<string|int, mixed>>>
Yields each row as an array.
getEach()  : Generator<string|int, Row>
Yield each row as an instance of Row.
getEachAsArray()  : Generator<string|int, array<string|int, mixed>>
Yield each row as an array.
groupBy()  : GroupedDataFrame
join()  : self
joinEach()  : self
Joins in memory per batch; it is not governed by the join algorithm and takes no algorithm override.
limit()  : self
load()  : self
match()  : self
offset()  : self
Skip given number of rows from the beginning of the dataset.
onError()  : self
printRows()  : void
printSchema()  : void
rename()  : self
renameEach()  : self
repartition()  : self
rows()  : self
run()  : Report : null)
schema()  : Schema
select()  : self
sortBy()  : self
transform()  : self
Alias for DataFrame::with().
until()  : self
The difference between filter and until is that filter will keep filtering rows until extractors finish yielding rows. Until will send a STOP signal to the Extractor when the condition is not met.
void()  : self
with()  : self
withEntries()  : self
withEntry()  : self
write()  : self

Methods

aggregate()

public aggregate(Aggregations $aggregations[, null|GroupByAlgorithmBuilder $algorithm = null ]) : self
Parameters
$aggregations : Aggregations
$algorithm : null|GroupByAlgorithmBuilder = null

null defers to configuration; a builder pins the algorithm for this operation and skips any automatic choice

Tags
lazy
Return values
self

batchBy()

Merge/Split Rows yielded by Extractor into batches but keep those with common value in given column together.

public batchBy(Reference|string $column[, null|int<1, max> $minSize = null ]) : self

This works properly only on sorted datasets.

When minSize is not provided, batches will be created only when there is a change in value of the column. When minSize is provided, batches will be created only when there is a change in value of the column or when there are at least minSize rows in the batch.

Parameters
$column : Reference|string
  • column to group by (all rows with same value stay together)
$minSize : null|int<1, max> = null
  • optional minimum rows per batch for efficiency
Tags
lazy
throws
InvalidArgumentException
Return values
self

batchSize()

Merge/Split Rows yielded by Extractor into batches of given size.

public batchSize(int<-1, max> $size) : self

For example, when Extractor is yielding one row at time, this method will merge them into batches of given size before passing them to the next pipeline element. Similarly when Extractor is yielding batches of rows, this method will split them into smaller batches of given size.

In order to merge all Rows into a single batch use DataFrame::collect() method or set size to -1 or 0.

Parameters
$size : int<-1, max>
Tags
lazy
Return values
self

cache()

Start processing rows up to this moment and put each instance of Rows into previously defined cache.

public cache([null|string $id = null ][, int|null $cacheBatchSize = null ][, null|Cache $cache = null ]) : self

Cache type can be set through ConfigBuilder. By default everything is cached in system tmp dir.

Important: cache batch size might significantly improve performance when processing large amount of rows. Larger batch size will increase memory consumption but will reduce number of IO operations. When not set, the batch size is taken from the last DataFrame::batchSize() call.

Parameters
$id : null|string = null
$cacheBatchSize : int|null = null
$cache : null|Cache = null

reads of this cache must pass the same instance to from_cache()

Tags
lazy
throws
InvalidArgumentException
Return values
self

collect()

Before transforming rows, collect them and merge into single Rows instance.

public collect() : self

This might lead to memory issues when processing large amount of rows, use with caution.

Tags
lazy
Return values
self

collectRefs()

This method allows to collect references to all entries used in this pipeline.

public collectRefs(References $references) : self 
(new Flow())
  ->read(From::chain())
  ->collectRefs($refs = refs())
  ->run();
Parameters
$references : References
Tags
lazy
Return values
self

count()

public count() : int
Tags
trigger

Return total count of rows processed by this pipeline.

Return values
int

crossJoin()

public crossJoin(self $dataFrame[, string $prefix = '' ]) : self
Parameters
$dataFrame : self
$prefix : string = ''
Tags
lazy
Return values
self

drop()

Drop given entries.

public drop(string|Reference ...$entries) : self
Parameters
$entries : string|Reference
Tags
lazy
Return values
self

dropDuplicates()

public dropDuplicates(Reference|string ...$entries) : $this
Parameters
$entries : Reference|string
Tags
lazy
Return values
$this

duplicateRow()

public duplicateRow(mixed $condition, WithEntry ...$entries) : self
Parameters
$condition : mixed
$entries : WithEntry
Return values
self

fetch()

Be aware that fetch is not memory safe and will load all rows into memory.

public fetch([int|null $limit = null ]) : Rows

If you want to safely iterate over Rows use oe of the following methods:.

DataFrame::get() : \Generator DataFrame::getAsArray() : \Generator DataFrame::getEach() : \Generator DataFrame::getEachAsArray() : \Generator

Parameters
$limit : int|null = null
Tags
trigger
throws
InvalidArgumentException
Return values
Rows

forEach()

public forEach([null|callable(Rows $rows): void $callback = null ]) : void
Parameters
$callback : null|callable(Rows $rows): void = null
Tags
trigger

get()

Yields each row as an instance of Rows.

public get() : Generator<string|int, Rows>
Tags
trigger
Return values
Generator<string|int, Rows>

getAsArray()

Yields each row as an array.

public getAsArray() : Generator<string|int, array<string|int, array<string|int, mixed>>>
Tags
trigger
Return values
Generator<string|int, array<string|int, array<string|int, mixed>>>

getEach()

Yield each row as an instance of Row.

public getEach() : Generator<string|int, Row>
Tags
trigger
Return values
Generator<string|int, Row>

getEachAsArray()

Yield each row as an array.

public getEachAsArray() : Generator<string|int, array<string|int, mixed>>
Tags
trigger
Return values
Generator<string|int, array<string|int, mixed>>

offset()

Skip given number of rows from the beginning of the dataset.

public offset(int<0, max>|null $offset) : self

When $offset is null, nothing happens (no rows are skipped).

Performance Note: DataFrame must iterate through and process all skipped rows to reach the offset position. For large offsets, this can impact performance as the data source still needs to be read and processed up to the offset point.

Parameters
$offset : int<0, max>|null
Tags
lazy
throws
InvalidArgumentException
Return values
self

rename()

public rename(string $from, string $to) : self
Parameters
$from : string
$to : string
Tags
lazy
Return values
self

repartition()

public repartition(string|Reference $entry, string|Reference ...$entries) : self
Parameters
$entry : string|Reference
$entries : string|Reference
Tags
lazy

Shuffles the stream so every row sharing the given columns arrives in one batch. It does not write directories - that is declared on the loader, to_csv(...)->partitionBy('region').

Return values
self

run()

public run([null|callable(Rows $rows, FlowContext $context): void $callback = null ][, Analyze|bool $analyze = false ]) : Report : null)
Parameters
$callback : null|callable(Rows $rows, FlowContext $context): void = null
$analyze : Analyze|bool = false
  • when set run will return Report
Tags
trigger

When analyzing pipeline execution we can chose to collect various metrics through analyze()->with*() method

  • column statistics - analyze()->withColumnStatistics()
  • schema - analyze()->withSchema()
Return values
Report : null)

select()

public select(string|Reference ...$entries) : self
Parameters
$entries : string|Reference
Tags
lazy

Keep only given entries.

Return values
self

sortBy()

public sortBy(Reference|SortReferences|string $entries[, null|SortAlgorithmBuilder $algorithm = null ]) : self
Parameters
$entries : Reference|SortReferences|string

a single column is sorted by on its own

$algorithm : null|SortAlgorithmBuilder = null

null defers to configuration; a builder pins the algorithm for this operation and skips any automatic choice

Tags
lazy
Return values
self

until()

The difference between filter and until is that filter will keep filtering rows until extractors finish yielding rows. Until will send a STOP signal to the Extractor when the condition is not met.

public until(ScalarFunction $function) : self
Parameters
$function : ScalarFunction
Tags
lazy
Return values
self

void()

public void() : self
Tags
lazy

This method is useful mostly in development when you want to pause processing at certain moment without removing code. All operations will get processed up to this point, from here no rows are passed forward.

Return values
self

write()

public write(Loader $loader) : self
Parameters
$loader : Loader
Tags
lazy

Alias for ETL::load function.

Return values
self
On this page

Search results