flow php

UNIFIED DATA PROCESSING FRAMEWORK

composer require flow-php/etl ^0.10.0

Changelog

elephant
extract

Extracts

Read from various data sources.

arrow
transform

Transforms

Shape and optimize for your needs.

arrow
load

Loads

Store and secure in one of many available data sinks.

Modules

DSL Functions

DSL stands for Domain Specific Language. In the case of Flow, the DSL is used to define simple functions that can be used to transform data. Most of those functions are initializing a new instance of a class under the hood since Flow is fully object-oriented. Please look at the examples below to get a better understanding of how to use the DSL functions.

SCHEMA

feature array_schema

feature source 
array_schema(string $name, bool $empty, bool $nullable, ?Metadata $metadata) : Definition

feature enum_schema

feature source 
/**
 * @param class-string<\UnitEnum> $type
 */
enum_schema(string $name, string $type, bool $nullable, ?Metadata $metadata) : Definition

TYPE

feature structure_type

feature source 
/**
 * @param array<StructureElement> $elements
 */
structure_type(array $elements, bool $nullable) : StructureType

feature struct_type

feature source 
/**
 * @param array<StructureElement> $elements
 */
struct_type(array $elements, bool $nullable) : StructureType

feature type_enum

feature source 
/**
 * @param class-string<\UnitEnum> $class
 */
type_enum(string $class, bool $nullable) : EnumType

feature type_map

feature source 
type_map(ScalarType $key_type, Type $value_type, bool $nullable) : MapType

feature type_object

feature source 
/**
 * @param class-string $class
 */
type_object(string $class, bool $nullable) : ObjectType

feature type_structure

feature source 
/**
 * @param array<StructureElement> $elements
 */
type_structure(array $elements, bool $nullable) : StructureType

ENTRY

feature array_entry

feature source 
/**
 * @param array<mixed> $data
 */
array_entry(string $array, ?array $data) : ArrayEntry

feature json_object_entry

feature source 
/**
 * @throws InvalidArgumentException
 */
json_object_entry(string $name, array|string|null $data) : JsonEntry

DATA_FRAME

feature array_to_row

feature source 
/**
 * @param array<array<mixed>>|array<mixed|string> $data
 * @param array<Partition>|Partitions $partitions
 */
array_to_row(array $data, EntryFactory $entryFactory, Partitions|array $partitions, ?Schema $schema) : Row

feature array_to_rows

feature source 
/**
 * @param array<array<mixed>>|array<mixed|string> $data
 * @param array<Partition>|Partitions $partitions
 */
array_to_rows(array $data, EntryFactory $entryFactory, Partitions|array $partitions, ?Schema $schema) : Rows

feature col

feature source 
/**
 * An alias for `ref`.
 */
col(string $entry) : EntryReference

feature df

feature source 
/**
 * Alias for data_frame() : Flow.
 */
df(Config|ConfigBuilder|null $config) : Flow

feature filesystem_cache

feature source 
filesystem_cache(Path|string|null $cache_dir, Filesystem $filesystem, Serializer $serializer) : FilesystemCache

feature is_type

feature source 
/**
 * @param array<string|Type> $types
 * @param mixed $value
 */
is_type(array $types, ?mixed $value) : bool

feature join_on

feature source 
join_on(Comparison|array $comparisons, string $join_prefix) : Expression

EXTRACTOR

feature chunks_from

feature source 
/**
 * @param int<1, max> $chunk_size
 */
chunks_from(Extractor $extractor, int $chunk_size) : ChunkExtractor

feature from_array

feature source 
/**
 * @param iterable $array
 * @param null|Schema $schema - @deprecated use withSchema() method instead
 */
from_array(iterable $array, ?Schema $schema) : ArrayExtractor

feature from_cache

feature source 
/**
 * @param string $id - cache id from which data will be extracted
 * @param null|Extractor $fallback_extractor - extractor that will be used when cache is empty - @deprecated use withFallbackExtractor() method instead
 * @param bool $clear - clear cache after extraction - @deprecated use withClearOnFinish() method instead
 */
from_cache(string $id, ?Extractor $fallback_extractor, bool $clear) : CacheExtractor

feature from_sequence_date_period

feature source 
from_sequence_date_period(string $entry_name, DateTimeInterface $start, DateInterval $interval, DateTimeInterface $end, int $options) : SequenceExtractor

feature from_sequence_number

feature source 
from_sequence_number(string $entry_name, string|int|float $start, string|int|float $end, int|float $step) : SequenceExtractor

AGGREGATING_FUNCTION

SCALAR_FUNCTION

feature array_expand

feature source 
/**
 * Expands each value into entry, if there are more than one value, multiple rows will be created.
 * Array keys are ignored, only values are used to create new rows.
 *
 * Before:
 *   +--+-------------------+
 *   |id|              array|
 *   +--+-------------------+
 *   | 1|{"a":1,"b":2,"c":3}|
 *   +--+-------------------+
 *
 * After:
 *   +--+--------+
 *   |id|expanded|
 *   +--+--------+
 *   | 1|       1|
 *   | 1|       2|
 *   | 1|       3|
 *   +--+--------+
 */
array_expand(ScalarFunction $function, ArrayExpand $expand) : ArrayExpand

feature array_key_rename

feature source 
array_key_rename(ScalarFunction $ref, ScalarFunction|string $path, ScalarFunction|string $newName) : ArrayKeyRename

feature array_reverse

feature source 
array_reverse(ScalarFunction|array $function, ScalarFunction|bool $preserveKeys) : ArrayReverse

feature array_sort

feature source 
array_sort(ScalarFunction $function, ScalarFunction|Sort|null $sort_function, ScalarFunction|int|null $flags, ScalarFunction|bool $recursive) : ArraySort

feature array_unpack

feature source 
/**
 * Unpacks each element of an array into a new entry, using the array key as the entry name.
 *
 * Before:
 * +--+-------------------+
 * |id|              array|
 * +--+-------------------+
 * | 1|{"a":1,"b":2,"c":3}|
 * | 2|{"d":4,"e":5,"f":6}|
 * +--+-------------------+
 *
 * After:
 * +--+-----+-----+-----+-----+-----+
 * |id|arr.b|arr.c|arr.d|arr.e|arr.f|
 * +--+-----+-----+-----+-----+-----+
 * | 1|    2|    3|     |     |     |
 * | 2|     |     |    4|    5|    6|
 * +--+-----+-----+-----+-----+-----+
 */
array_unpack(ScalarFunction|array $array, ScalarFunction|array $skip_keys, ScalarFunction|string|null $entry_prefix) : ArrayUnpack

feature between

feature source 
between(?mixed $value, ?mixed $lower_bound, ?mixed $upper_bound, ScalarFunction|Boundary $boundary) : Between

feature call_method

feature source 
/**
 * @param array<mixed> $params
 */
call_method(object $object, ScalarFunction|string $method, array $params) : CallMethod

feature cast

feature source 
cast(?mixed $value, ScalarFunction|Type|string $type) : Cast

feature combine

feature source 
combine(ScalarFunction|array $keys, ScalarFunction|array $values) : Combine

feature entry

feature source 
/**
 * An alias for `ref`.
 */
entry(string $entry) : EntryReference

feature number_format

feature source 
number_format(ScalarFunction|int|float $value, ScalarFunction|int $decimals, ScalarFunction|string $decimal_separator, ScalarFunction|string $thousands_separator) : NumberFormat

feature regex

feature source 
regex(ScalarFunction|string $pattern, ScalarFunction|string $subject, ScalarFunction|int $flags, ScalarFunction|int $offset) : Regex

feature regex_all

feature source 
regex_all(ScalarFunction|string $pattern, ScalarFunction|string $subject, ScalarFunction|int $flags, ScalarFunction|int $offset) : RegexAll

feature regex_match

feature source 
regex_match(ScalarFunction|string $pattern, ScalarFunction|string $subject, ScalarFunction|int $flags, ScalarFunction|int $offset) : RegexMatch

feature regex_match_all

feature source 
regex_match_all(ScalarFunction|string $pattern, ScalarFunction|string $subject, ScalarFunction|int $flags, ScalarFunction|int $offset) : RegexMatchAll

feature regex_replace

feature source 
regex_replace(ScalarFunction|string $pattern, ScalarFunction|string $replacement, ScalarFunction|string $subject, ScalarFunction|int|null $limit) : RegexReplace

feature round

feature source 
round(ScalarFunction|int|float $value, ScalarFunction|int $precision, ScalarFunction|int $mode) : Round

feature sanitize

feature source 
sanitize(ScalarFunction|string $value, ScalarFunction|string $placeholder, ScalarFunction|int|null $skipCharacters) : Sanitize

feature split

feature source 
split(ScalarFunction|string $value, ScalarFunction|string $separator, ScalarFunction|int $limit) : Split

feature sprintf

feature source 
sprintf(ScalarFunction|string $format, ScalarFunction|string|int|float|null $args) : Sprintf

feature to_date

feature source 
to_date(?mixed $ref, ScalarFunction|string $format, ScalarFunction|DateTimeZone $timeZone) : ToDate

feature to_date_time

feature source 
to_date_time(?mixed $ref, ScalarFunction|string $format, ScalarFunction|DateTimeZone $timeZone) : ToDateTime

feature to_timezone

feature source 
to_timezone(ScalarFunction|DateTimeInterface $value, ScalarFunction|DateTimeZone|string $timeZone) : ToTimeZone

feature when

feature source 
when(?mixed $condition, ?mixed $then, ?mixed $else) : When

WINDOW_FUNCTION

COMPARISON

feature equal

feature source 
equal(Reference|string $left, Reference|string $right) : Equal

LOADER

feature to_array

feature source 
/**
 * Convert rows to an array and store them in passed array variable.
 *
 * @param-out array<array<mixed>> $array
 */
to_array(array $array) : ArrayLoader

feature to_output

feature source 
to_output(int|bool $truncate, Output $output, Formatter $formatter, SchemaFormatter $schemaFormatter) : StreamLoader

feature to_stderr

feature source 
to_stderr(int|bool $truncate, Output $output, Formatter $formatter, SchemaFormatter $schemaFormatter) : StreamLoader

feature to_stdout

feature source 
to_stdout(int|bool $truncate, Output $output, Formatter $formatter, SchemaFormatter $schemaFormatter) : StreamLoader

feature to_stream

feature source 
to_stream(string $uri, int|bool $truncate, Output $output, string $mode, Formatter $formatter, SchemaFormatter $schemaFormatter) : StreamLoader

Contributors

Join us on GitHub external resource
norberttech stloyd tomaszhanc DawidSajdak owsiakl flavioheleno rzarno szepeviktor mleczakm alexislefebvre scyzoryck xaviermarchegay peter279k Wirone jpiatko jguittard voku drupol Wiktor6 inmanturbo
scroll back to top