Source Feed

A Feed is a runtime platform component responsible for interpreting the ETL query defined as the project source and resolving the requested dataset using the linked storage system.


The Feed concept is based on two main principles:

  1. A DSL-interpreting Reader acting as an adapter between the storage layer and the pipeline.

  2. A content resolver using an explicit mapping of the published schema catalogs to the hosted data sources effectively matching the logical schemas with actual data.

Content resolution takes place in the scope of the DSL parsing as part of the Reader routine simply by visiting and replacing the matched DSL Sources/Features with the mapped terms declared already using the parser-target semantic.

When launching the pipeline, ForML runner expands the Feed into one or more initial tasks within the assembled workflow making it a native part of the final DAG to be executed.

The core Feed API looks as follows:

alias of Callable[[dsl.Statement, Optional[layout.Entry]], layout.Tabular]


Abstract base class for data source feed providers.

It integrates the concept of a DSL-based Reader provided using the producer() method or by overriding the inner .Reader class and a content resolver with its abstract parts represented by the sources and features properties.

The need for implementing the content resolver mapping specifically for each particular platform makes it more difficult to setup Feed providers using just the parametric configuration and might end up requiring to actually implement the Feed (or at least the final resolver part) explicitly as a code.


Feeds need to be serializable!

classmethod producer(sources: Mapping[dsl.Source, parser.Source], features: Mapping[dsl.Feature, parser.Feature], **kwargs: Any) io.Producer[source]

Producer factory method.

A Producer is a generic callable interface most typically represented using the implementation whose task is to parse the provided DSL query and resolve it using its linked storage.

Unless overloaded, the method returns an instance of cls.Reader (which might be easier to extend without needing to overload this method).

sources: Mapping[dsl.Source, parser.Source]

Source mappings to be used by the reader (see sources).

features: Mapping[dsl.Feature, parser.Feature]

Column mappings to be used by the reader (see features).

**kwargs: Any

Optional reader keyword arguments.


Producer instance.

abstract property sources : Mapping[dsl.Source, parser.Source]

The main part of the content resolver providing the Source mappings.

This way the Feed is advertising the available datasets represented using their published schemas logically mapped to the hosted data sources specified using the parser-specific semantics.

A Source is a DSL concept representing anything that can be queried as a data source.


Sources mapping.


Using a parser with SQLAlchemy semantics, an example of the mapping might look like this:

return {
    schema.Titanic: sqlalchemy.table('titanic'),
    foo.Bar.join(foo.Baz, == sqlalchemy.table('foobar'),

Note the capability of mapping a complex query to a denormalized dataset.

property features : Mapping[dsl.Feature, parser.Feature]

The minor part of the content resolver providing the optional Feature mappings.

Optional mapping of individual Features to their hosted representation using the parser-specific semantic.

A Feature is a DSL concept representing anything that can be projected to a data column.


Features mapping.

class Mapping[dsl.Source, parser.Source], features: Mapping[dsl.Feature, parser.Feature], **kwargs: Any)

Generic reader base class matching the Feed producer interface.

It is a low-level input component responsible for parsing a generic data request in the form of a DSL query (the parser() method) and based on it retrieving the actual data from its supported storage technology and its specific data format (the read() method).

Reader can operate in two possible modes depending on the input parameters:

  • extraction - when called just using a query without the input entry parameter, it simply executes the (parsed) query against the backend storage.

  • augmentation - if the entry value is provided, it is interpreted as the actual source to be returned but potentially incomplete in terms of the expected schema; in which case the reader is supposed to just complete the partial data to match the query schema.


Implement the augmentation mode.

abstract classmethod parser(sources: Mapping[dsl.Source, parser.Source], features: Mapping[dsl.Feature, parser.Feature]) parser.Visitor

Parser factory method.

The parser instance must be able to convert a DSL query into storage-native instructions compatible with the read() method.

During parsing, the provided sources/features mappings are supposed to be used for the content resolution.

sources: Mapping[dsl.Source, parser.Source]

Source mappings to be used by the parser.

features: Mapping[dsl.Feature, parser.Feature]

Feature mappings to be used by the parser.


Parser instance.

classmethod format(schema: dsl.Source.Schema, data: layout.Native) layout.Tabular

Convert the storage-native data into the required layout.Tabular format.

schema: dsl.Source.Schema

Data schema.

data: layout.Native

Input data.


Data formatted into the layout.Tabular format.

abstract classmethod read(statement: parser.Source, **kwargs: Any) layout.Native

Perform the read operation using the given storage-native statement.

statement: parser.Source

Read instructions in the storage-native syntax.

**kwargs: Any

Optional reader keyword arguments (as given to the constructor).


Raw data provided by the reader.

For reference, several existing Reader implementations can be found under the forml.provider.feed.reader package:


SQLAlchemy based reader.

Contextual Feed Selection

Unlike the other provider types which explicitly nominate exactly one instance each before launching, feeds go through a more dynamic process of selecting the most suitable candidate in the context of the actual data query.

For this purpose, ForML uses the io.Importer class:

class*feeds: setup.Feed | str | io.Feed)[source]

Pool of (possibly) lazily instantiated feeds.

The pool is used to select the most suitable feed instance capable of resolving the particular DSL query (in terms of providing data sources for all the involved schemas). Feed instances can have a static priority assigned in which case the first feed with the highest priority capable of providing the data is returned.

If configured without any explicit instances, all the feeds registered in the provider cache are pooled.


This logic should be extended to also probe the available data range so that a feed without the expected data range is not prioritized over another feed that has the range but has a lower absolute priority.

match(source: dsl.Source) io.Feed[source]

Select a feed instance that can supply data for the given source.

source: dsl.Source

Any DSL Source representing the data request.


Feed that’s able to provide data for the given request.


forml.MissingError – If no feed can provide the requested data.

Custom Feed Setup

Existing generic Feed implementations can be configured as any other provider types. The strong deployment-specific character of the content resolver setup (explicit mapping of the published schemas and the hosted data sources) might, however, require to declare bespoke Feed providers using not just parametric configuration but rather directly as a non-generic code.

For more details, see the custom provider setup instructions.

Feed Providers

Feed providers can be configured within the runtime platform setup using the [FEED.*] sections.

The available implementations are:


Generic SQL feed based on SQLAlchemy.


Lightweight feed for pulling data from multiple simple origins.

External Providers


ForML feed providing access to a number of public datasets.