Download the PHP package code-rhapsodie/dataflow-bundle without Composer

Code Rhapsodie Dataflow Bundle

DataflowBundle is a bundle for Symfony 3.4+ providing an easy way to create import / export dataflow.

Dataflow uses a linear generic workflow in three parts:

• one reader
• any number of steps that can be synchronous or asynchronous
• one or more writers

The reader can read data from anywhere and return data row by row. Each step processes the current row data. The steps are executed in the order in which they are added. And, one or more writers save the row anywhere you want.

As the following schema shows, you can define more than one dataflow:

Features

• Define and configure a Dataflow
• Run the Job scheduled
• Run one Dataflow from the command line
• Define the schedule for a Dataflow from the command line
• Enable/Disable a scheduled Dataflow from the command line
• Display the list of scheduled Dataflow from the command line
• Display the result for the last Job for a Dataflow from the command line
• Work with multiple Doctrine DBAL connections

Installation

Security notice: Symfony 4.x is not supported before 4.1.12, see https://github.com/advisories/GHSA-pgwj-prpq-jpc2

Add the dependency

To install this bundle, run this command :

Suggest

You can use the generic readers, writers and steps from PortPHP.

For the writers, you must use the adapter CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter like this:

Register the bundle

Add CodeRhapsodie\DataflowBundle\CodeRhapsodieDataflowBundle::class => ['all' => true], in the config/bundles.php file.

Like this:

Update the database

This bundle uses Doctrine DBAL to store Dataflow schedule into the database table (cr_dataflow_scheduled) and jobs (cr_dataflow_job).

If you use Doctrine Migration Bundle or Phinx or Kaliop Migration Bundle or whatever, you can add a new migration with the generated SQL query from this command:

If you have already the tables, you can add a new migration with the generated update SQL query from this command:

Configuration

By default, the Doctrine DBAL connection used is default. You can configure the default connection. Add this configuration into your Symfony configuration:

By default, the logger service will be used to log all exceptions and custom messages. If you want to use another logger, like a specific Monolog handler, Add this configuration:

Messenger mode

Dataflow can delegate the execution of its jobs to the Symfony messenger component, if available. This allows jobs to be executed concurrently by workers instead of sequentially.

To enable messenger mode:

You also need to route Dataflow messages to the proper transport:

Exceptions mode

Dataflow can save exceptions in any filesystem you want. This allows dataflow to save exceptions in filesystem instead of the database You have to install league/flysystem.

To enable exceptions mode:

Define a dataflow type

This bundle uses a fixed and simple workflow structure in order to let you focus on the data processing logic part of your dataflow.

A dataflow type defines the different parts of your dataflow. A dataflow is made of:

• exactly one Reader
• any number of Steps
• one or more Writers

Dataflow types can be configured with options.

A dataflow type must implement CodeRhapsodie\DataflowBundle\DataflowType\DataflowTypeInterface.

To help with creating your dataflow types, an abstract class CodeRhapsodie\DataflowBundle\DataflowType\AbstractDataflowType is provided, allowing you to define your dataflow through a handy builder CodeRhapsodie\DataflowBundle\DataflowType\DataflowBuilder.

This is an example to define one class DataflowType:

Dataflow types must be tagged with coderhapsodie.dataflow.type.

If you're using Symfony auto-configuration for your services, this tag will be automatically added to all services implementing DataflowTypeInterface.

Otherwise, manually add the tag coderhapsodie.dataflow.type in your dataflow type service configuration:

Use options for your dataflow type

The AbstractDataflowType can help you define options for your Dataflow type.

Add this method in your DataflowType class:

With this configuration, the option fileName is required. For an advanced usage of the option resolver, read the Symfony documentation.

For asynchronous management, AbstractDataflowType come with two default options :

• loopInterval : default to 0. Update this interval if you wish customise the tick loop duration.
• emitInterval : default to 0. Update this interval to have a control when reader must emit new data in the flow pipeline.

Logging

All exceptions will be caught and written in the logger. If you want to add custom messages in the log, you can inject the logger in your readers / steps / writers. If your DataflowType class extends AbstractDataflowType, the logger is accessible as $this->logger.

When using the code-rhapsodie:dataflow:run-pending command, this logger will also be used to save the log in the corresponding job in the database.

Check if your DataflowType is ready

Execute this command to check if your DataflowType is correctly registered:

The result is like this:

Readers

Readers provide the dataflow with elements to import / export. Usually, elements are read from an external resource ( file, database, webservice, etc).

A Reader can be any iterable.

The only constraint on the returned elements typing is that they cannot be false.

The reader can be a generator like this example :

You can set up this reader as follows:

Steps

Steps are operations performed on the elements before they are handled by the Writers. Usually, steps are either:

• converters, that alter the element
• filters, that conditionally prevent further operations on the element
• generators, that can include asynchronous operations

A Step can be any callable, taking the element as its argument, and returning either:

• the element, possibly altered
• false, if no further operations should be performed on this element

A few examples:

Note : you can ensure writing order for asynchronous operations if all steps are scaled at 1 factor.

Writers

Writers perform the actual import / export operations.

A Writer must implement CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface. As this interface is not compatible with Port\Writer, the adapter CodeRhapsodie\DataflowBundle\DataflowType\Writer\PortWriterAdapter is provided.

This example show how to use the predefined PhpPort Writer :

Or your own Writer:

CollectionWriter

If you want to write multiple items from a single item read, you can use the generic CollectionWriter. This writer will iterate over any iterable it receives, and pass each item from that collection to your own writer that handles single items.

DelegatorWriter

If you want to call different writers depending on what item is read, you can use the generic DelegatorWriter.

As an example, let's suppose our items are arrays with the first entry being either product or order. We want to use a different writer based on that value.

First, create your writers implementing DelegateWriterInterface (this interface extends WriterInterface so your writers can still be used without the DelegatorWriter).

Then, configure your DelegatorWriter and add it to your dataflow type.

During execution, the DelegatorWriter will simply pass each item received to its first delegate (in the order those were added) that supports it. If no delegate supports an item, an exception will be thrown.

Queue

All pending dataflow job processes are stored in a queue into the database.

Add this command into your crontab for execute all queued jobs:

Commands

Several commands are provided to manage schedules and run jobs.

code-rhapsodie:dataflow:run-pending Executes job in the queue according to their schedule.

When messenger mode is enabled, jobs will still be created according to their schedule, but execution will be handled by the messenger component instead.

code-rhapsodie:dataflow:schedule:list Display the list of dataflows scheduled.

code-rhapsodie:dataflow:schedule:change-status Enable or disable a scheduled dataflow

code-rhapsodie:dataflow:schedule:add Add the schedule for a dataflow.

code-rhapsodie:dataflow:job:show Display the last result of a job.

code-rhapsodie:dataflow:execute Let you execute one dataflow job.

code-rhapsodie:dataflow:dump-schema Generates schema create / update SQL queries

code-rhapsodie:dataflow:job:set-crashed Jobs that have been in the "running" status for too long will be set in the "crashed" status.

code-rhapsodie:dataflow:job:cleanup Remove old completed or crashed jobs

Work with many databases

All commands have a --connection option to define what Doctrine DBAL connection to use during execution.

Example:

This command uses the default DBAL connection to generate all schema update queries.

To execute all pending job for a specific connection use:

Issues and feature requests

Please report issues and request features at https://github.com/code-rhapsodie/dataflow-bundle/issues.

Please note that only the last release of the 4.x and the 5.x versions of this bundle are actively supported.

Contributing

Contributions are very welcome. Please see CONTRIBUTING.md for details. Thanks to everyone who has contributed already.

License

This package is licensed under the MIT license.