frictionlessdata/datapackage-pipelines repository overview

⁠Datapackage Pipelines

⁠The Basics

⁠What is it?

datapackage-pipelines is a framework for declarative stream-processing of tabular data. It is built upon the concepts and tooling of the Frictionless Data project.

⁠Pipelines

The basic concept in this framework is the pipeline.

A pipeline has a list of processing steps, and it generates a single data package as its output. Each step is executed in a processor and consists of the following stages:

• Modify the data package descriptor - For example: add metadata, add or remove resources, change resources' data schema etc.
• Process resources - Each row of each resource is processed sequentially. The processor can drop rows, add new ones or modify their contents.
• Return stats - If necessary, the processor can report a dictionary of data which will be returned to the user when the pipeline execution terminates. This can be used, for example, for calculating quality measures for the processed data.

Not every processor needs to do all of these. In fact, you would often find each processing step doing only one of these.

⁠pipeline-spec.yaml file

Pipelines are defined in a declarative way, and not in code. One or more pipelines can be defined in a pipeline-spec.yaml file. This file specifies the list of processors (referenced by name) and the execution parameters for each of the processors.

Here's an example of a pipeline-spec.yaml file:

worldbank-co2-emissions:
  title: CO2 emission data from the World Bank
  description: Data per year, provided in metric tons per capita.
  environment:
    DEBUG: true
  pipeline:
    -
      run: update_package
      parameters:
        name: 'co2-emissions'
        title: 'CO2 emissions (metric tons per capita)'
        homepage: 'http://worldbank.org/'
    -
      run: load
      parameters:
        from: "http://api.worldbank.org/v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel"
        name: 'global-data'
        format: xls
        headers: 4
    -
      run: set_types
      parameters:
         resources: global-data
         types:
           "[12][0-9]{3}":
              type: number
    -
      run: dump_to_zip
      parameters:
          out-file: co2-emissions-wb.zip

In this example we see one pipeline called worldbank-co2-emissions. Its pipeline consists of 4 steps:

• update_package: This is a library processor (see below), which modifies the data-package's descriptor (in our case: the initial, empty descriptor) - adding name, title and other properties to the datapackage.
• load: This is another library processor, which loads data into the data-package. This resource has a name and a from property, pointing to the remote location of the data.
• set_types: This processor assigns data types to fields in the data. In this example, field headers looking like years will be assigned the number type.
• dump_to_zip: Create a zipped and validated datapackage with the provided file name.

Also, we have provided some metadata:

• title: Title of a pipeline
• description: Description of a pipeline
• environment: Dictionary of environment variables to be set for all the pipeline's steps. For examples, it can be used to change the behaviour of the underlaying requests library - https://requests.readthedocs.io/en/master/user/advanced/#ssl-cert-verification⁠

⁠Mechanics

An important aspect of how the pipelines are run is the fact that data is passed in streams from one processor to another. If we get "technical" here, then each processor is run in its own dedicated process, where the datapackage is read from its stdin and output to its stdout. The important thing to note here is that no processor holds the entire data set at any point.

This limitation is by design - to keep the memory and disk requirements of each processor limited and independent of the dataset size.

⁠Quick Start

First off, create a pipeline-spec.yaml file in your current directory. You can take the above file if you just want to try it out.

Then, you can either install datapackage-pipelines locally - note that Python 3.6 or higher is required due to use of Type Hinting⁠ and advanced asyncio use:

$ pip install datapackage-pipelines

You should now be able to use the dpp command:

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1" 200 308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1" 200 308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1" 200 308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1" 200 308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

Alternatively, you could use our Docker⁠ image:

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

⁠The Command Line Interface - dpp

Running a pipeline from the command line is done using the dpp tool.

Running dpp without any argument, will show the list of available pipelines. This is done by scanning the current directory and its subdirectories, searching for pipeline-spec.yaml files and extracting the list of pipeline specifications described within.

Each pipeline has an identifier, composed of the path to the pipeline-spec.yaml file and the name of the pipeline, as defined within that description file.

In order to run a pipeline, you use dpp run <pipeline-id>.

You can also use dpp run all for running all pipelines and dpp run dirty to run the just the dirty pipelines (more on that later on).

⁠Deeper look into pipelines

⁠Processor Resolution

As previously seen, processors are referenced by name.

This name is, in fact, the name of a Python script containing the processing code (minus the .py extension). When trying to find where is the actual code that needs to be executed, the processor resolver will search in these predefined locations:

• First of all, it will try to find a custom processor with that name in the directory of the pipeline-spec.yaml file. Processor names support the dot notation, so you could write mycode.custom_processor and it will try to find a processor named custom_processor.py in the mycode directory, in the same path as the pipeline spec file. For this specific resolving phase, if you would write ..custom_processor it will try to find that processor in the parent directory of the pipeline spec file. (read on for instructions on how to write custom processors)
• In case the processor name looks like myplugin.somename, it will try to find a processor named somename in the myplugin plugin. That is - it will see if there's an installed plugin which is called myplugin, and if so, whether that plugin publishes a processor called somename (more on plugins below).
• If no processor was found until this point, it will try to search for this processor in the processor search path. The processor search path is taken from the environment variable DPP_PROCESSOR_PATH. Each of the : separated paths in the path is considered as a possible starting point for resolving the processor.
• Finally, it will try to find that processor in the Standard Processor Library which is bundled with this package.

⁠Excluding directories form scanning for pipeline specs

By default .* directories are excluded from scanning, you can add additional directory patterns for exclusion by creating a .dpp_spec_ignore file at the project root. This file has similar syntax to .gitignore and will exclude directories from scanning based on glob pattern matching.

For example, the following file will ignore test* directories including inside subdirectories and /docs directory will only be ignored at the project root directory

test*
/docs

⁠Caching

By setting the cached property on a specific pipeline step to True, this step's output will be stored on disk (in the .cache directory, in the same location as the pipeline-spec.yaml file).

Rerunning the pipeline will make use of that cache, thus avoiding the execution of the cached step and its precursors.

Internally, a hash is calculated for each step in the pipeline - which is based on the processor's code, it parameters and the hash of its predecessor. If a cache file exists with exactly the same hash as a specific step, then we can remove it (and its predecessors) and use that cache file as an input to the pipeline

This way, the cache becomes invalid in case the code or execution parameters changed (either for the cached processor or in any of the preceding processors).

⁠Dirty tasks and keeping state

The cache hash is also used for seeing if a pipeline is "dirty". When a pipeline completes executing successfully, dpp stores the cache hash along with the pipeline id. If the stored hash is different than the currently calculated hash, it means that either the code or the execution parameters were modified, and that the pipeline needs to be re-run.

dpp works with two storage backends. For running locally, it uses a python sqlite DB to store the current state of each running task, including the last result and cache hash. The state DB file is stored in a file named .dpp.db in the same directory that dpp is being run from.

For other installations, especially ones using the task scheduler, it is recommended to work with the Redis backend. In order to enable the Redis connection, simply set the DPP_REDIS_HOST environment variable to point to a running Redis instance.

⁠Pipeline Dependencies

You can declare that a pipeline is dependent on another pipeline or datapackage. This dependency is considered when calculating the cache hashes of a pipeline, which in turn affect the validity of cache files and the "dirty" state:

• For pipeline dependencies, the hash of that pipeline is used in the calculation
• For datapackage dependencies, the hash property in the datapackage is used in the calculation

If the dependency is missing, then the pipeline is marked as 'unable to be executed'.

Declaring dependencies is done by a dependencies property to a pipeline definition in the pipeline-spec.yaml file. This property should contain a list of dependencies, each one is an object with the following formats:

• A single key named pipeline whose value is the pipeline id to depend on
• A single key named datapackage whose value is the identifier (or URL) for the datapackage to depend on

Example:

cat-vs-dog-populations:
  dependencies:
    -
      pipeline: ./geo/region-areal
    -
      datapackage: http://pets.net/data/dogs-per-region/datapackage.json
    -
      datapackage: http://pets.net/data/dogs-per-region
  ...

⁠Validating

Each processor's input is automatically validated for correctness:

• The datapackage is always validated before being passed to a processor, so there's no possibility for a processor to modify a datapackage in a way that renders it invalid.

• Data is not validated against its respective JSON Table Schema, unless explicitly requested by setting the validate flag to True in the step's info. This is done for two main reasons:

  • Performance wise, validating the data in every step is very CPU intensive
  • In some cases you modify the schema in one step and the data in another, so you would only like to validate the data once all the changes were made

  In any case, when using the set_types standard processor, it will validate and transform the input data with the new types..

⁠Dataflows integration

Dataflows⁠ is the successor of datapackage-pipelines and provides a more Pythonic interface to running pipelines. You can integrate dataflows within pipeline specs using the flow attribute instead of run. For example, given the following flow file, saved under my-flow.py:

from dataflows import Flow, dump_to_path, load, update_package

def flow(parameters, datapackage, resources, stats):
    stats['multiplied_fields'] = 0

    def multiply(field, n):
        def step(row):
            row[field] = row[field] * n
            stats['multiplied_fields'] += 1
        return step

    return Flow(update_package(name='my-datapackage'),
                multiply('my-field', 2))

And a pipeline-spec.yaml in the same directory:

my-flow:
  pipeline:
  - run: load_resource
    parameters:
      url: http://example.com/my-datapackage/datapackage.json
      resource: my-resource
  - flow: my-flow
  - run: dump_to_path

You can run the pipeline using dpp run my-flow.

⁠The Standard Processor Library

A few built in processors are provided with the library.

⁠update_package

Adds meta-data to the data-package.

Parameters:

Any allowed property (according to the spec⁠) can be provided here.

Example:

- run: update_package
  parameters:
    name: routes-to-mordor
    license: CC-BY-SA-4
    author: Frodo Baggins <[email protected]>
    contributors:
      - samwise gamgee <[email protected]>

⁠update_resource

Adds meta-data to the resource.

Parameters:

• resources
  • A name of a resource to operate on
  • A regular expression matching resource names
  • A list of resource names
  • The index of the resource in the package
  • if omitted indicates operation should be done on all resources
• metadata - A dictionary containing any allowed property (according to the spec⁠).

Example:

- run: update_resource
  parameters:
    resources: ['resource1']
    metadata:
      path: 'new-path.csv'

⁠load

Loads data into the package, infers the schema and optionally casts values.

Parameters:

• from - location of the data that is to be loaded. This can be either:
  • a local path (e.g. /path/to/the/data.csv)
  • a remote URL (e.g. https://path.to/the/data.csv⁠)
  • Other supported links, based on the current support of schemes and formats in tabulator⁠
  • a local path or remote URL to a datapackage.json file (e.g. https://path.to/data_package/datapackage.json⁠)
  • a reference to an environment variable containing the source location, in the form of env://ENV_VAR
  • a tuple containing (datapackage_descriptor, resources_iterator)
• resources - optional, relevant only if source points to a datapackage.json file or datapackage/resource tuple. Value should be one of the following:
  • Name of a single resource to load
  • A regular expression matching resource names to load
  • A list of resource names to load
  • 'None' indicates to load all resources
  • The index of the resource in the package
• validate - Should data be casted to the inferred data-types or not. Relevant only when not loading data from datapackage.
• other options - based on the loaded file, extra options (e.g. sheet for Excel files etc., see the link to tabulator above)

⁠printer

Just prints whatever it sees. Good for debugging.

Parameters:

• num_rows - modify the number of rows to preview, printer will print multiple samples of this number of rows from different places in the stream
• last_rows - how many of the last rows in the stream to print. optional, defaults to the value of num_rows
• fields - optional, list of field names to preview
• resources - optional, allows to limit the printed resources, same semantics as load processor resources argument

⁠set_types

Sets data types and type options to fields in streamed resources, and make sure that the data still validates with the new types.

This allows to make modifications to the existing table schema, and usually to the default schema from stream_remote_resources.

Parameters:

• resources - Which resources to modify. Can be:

  • List of strings, interpreted as resource names to stream
  • String, interpreted as a regular expression to be used to match resource names

  If omitted, all resources in datapackage are streamed.

• regex - if set to False field names will be interpreted as strings not as regular expressions (True by default)

• types - A map between field names and field definitions.

  • field name is either simply the name of a field, or a regular expression matching multiple fields.
  • field definition is an object adhering to the JSON Table Schema spec⁠. You can use null instead of an object to remove a field from the schema.

Example:

- run: add_resources
  parameters:
    name: example-resource
    url: http://example.com/my-csv-file.csv
    encoding: "iso-8859-2"
- run: stream_remote_resources
- run: set_types
  parameters:
    resources: example-resource
    types:
      age:
        type: integer
      "yearly_score_[0-9]{4}":
        type: number
      "date of birth":
        type: date
        format: "%d/%m/%Y"
      "social security number": null

⁠load_metadata

Loads metadata from an existing data-package.

Parameters:

Loads the metadata from the data package located at url.

All properties of the loaded datapackage will be copied (except the resources)

Example:

- run: load_metadata
  parameters:
    url: http://example.com/my-datapackage/datapackage.json

⁠load_resource

Loads a tabular resource from an existing data-package.

Parameters:

Loads the resource specified in the resource parameter from the data package located at url. All properties of the loaded resource will be copied - path and schema included.

• url - a URL pointing to the datapackage in which the required resource resides

• resource - can be

  • List of strings, interpreted as resource names to load
  • String, interpreted as a regular expression to be used to match resource names
  • an integer, indicating the index of the resource in the data package (0-based)

• limit-rows - if provided, will limit the number of rows fetched from the source. Takes an integer value which specifies how many rows of the source to stream.

• log-progress-rows - if provided, will log the loading progress. Takes an integer value which specifies the number of rows interval at which to log the progress.

• stream - if provided and is set to false, then the resource will be added to the datapackage but not streamed.

• resources - can be used instead of resource property to support loading resources and modify the output resource metadata

  • Value is a dict containing mapping between source resource name to load and dict containing descriptor updates to apply to the loaded resource

• required - if provided and set to false, will not fail if datapackage is not available or resource is missing

Example:

- run: load_resource
  parameters:
    url: http://example.com/my-datapackage/datapackage.json
    resource: my-resource
- run: load_resource
  parameters:
    url: http://example.com/my-other-datapackage/datapackage.json
    resource: 1
- run: load_resource
  parameters:
    url: http://example.com/my-datapackage/datapackage.json
    resources:
      my-resource:
        name: my-renamed-resource
        path: my-renamed-resource.csv

⁠concatenate

Concatenates a number of streamed resources and converts them to a single resource.

Parameters:

• sources - Which resources to concatenate. Same semantics as resources in stream_remote_resources.

  If omitted, all resources in datapackage are concatenated.

  Resources to concatenate must appear in consecutive order within the data-package.

• target - Target resource to hold the concatenated data. Should define at least the following properties:

  • name - name of the resource
  • path - path in the data-package for this file.

  If omitted, the target resource will receive the name concat and will be saved at data/concat.csv in the datapackage.

• fields - Mapping of fields between the sources and the target, so that the keys are the target field names, and values are lists of source field names.

  This mapping is used to create the target resources schema.

  Note that the target field name is always assumed to be mapped to itself.

Example:

- run: concatenate
  parameters:
    target:
      name: multi-year-report
      path: data/multi-year-report.csv
    sources: 'report-year-20[0-9]{2}'
    fields:
      activity: []
      amount: ['2009_amount', 'Amount', 'AMOUNT [USD]', '$$$']

In this example we concatenate all resources that look like report-year-<year>, and output them to the multi-year-report resource.

The output contains two fields:

• activity , which is called activity in all sources
• amount, which has varying names in different resources (e.g. Amount, 2009_amount, amount etc.)

⁠join

Joins two streamed resources.

"Joining" in our case means taking the target resource, and adding fields to each of its rows by looking up data in the source resource.

A special case for the join operation is when there is no target stream, and all unique rows from the source are used to create it. This mode is called deduplication mode - The target resource will be created and deduplicated rows from the source will be added to it.

Parameters:

• source - information regarding the source resource
  • name - name of the resource
  • key - One of
    • List of field names which should be used as the lookup key
    • String, which would be interpreted as a Python format string used to form the key (e.g. {<field_name_1>}:{field_name_2})
  • delete - delete from data-package after joining (False by default)
• target - Target resource to hold the joined data. Should define at least the following properties:
  • name - as in source
  • key - as in source, or null for creating the target resource and performing deduplication.
• fields - mapping of fields from the source resource to the target resource. Keys should be field names in the target resource. Values can define two attributes:

  • name - field name in the source (by default is the same as the target field name)

  • aggregate - aggregation strategy (how to handle multiple source rows with the same key). Can take the following options:

    • sum - summarise aggregated values. For numeric values