Datamorph

Datamorph is a Laravel package that allows you to create and run Flow PHP ETL (Extract, Transform, Load) pipelines in a structured and extensible way. This documentation will guide you through the installation, configuration, and usage of the package.

Table of Contents

• Installation
• Concepts
• Configuration
• Creating an ETL Pipeline
• Running an ETL Pipeline
• Hooks
• Concrete Examples

Installation

Install the package via Composer:

composer require pollora/datamorph

Concepts

Datamorph is built around three main components:

1. Extractors: Retrieve data from various sources (databases, APIs, files, etc.)
2. Transformers: Transform the retrieved data according to your needs
3. Loaders: Load the transformed data to their final destination

These three components are orchestrated in a Pipeline that also manages Hooks that allow you to intervene at different stages of the process.

Configuration

Publish the configuration file:

php artisan vendor:publish --tag=datamorph-config

This will create a config/datamorph.php file where you can configure your ETL pipelines:

return [
'pipelines' => [
'stock' => [
'hooks' => [
'before_extract' => [
App\ETL\Stock\Hooks\BeforeStockExtract::class,
],
'after_extract' => [
// Hooks to execute after extraction
],
'before_transform' => [
// Hooks to execute before transformation
],
'after_transform' => [
// Hooks to execute after transformation
],
'before_load' => [
// Hooks to execute before loading
],
'after_load' => [
// Hooks to execute after loading
],
'before_run' => [
App\ETL\Stock\Hooks\BeforeStockRun::class,
],
'after_run' => [
// Hooks to execute after complete execution
],
],
],
// Other pipelines...
],
];

Creating an ETL Pipeline

Automatic File Generation

Datamorph includes an Artisan command that automatically generates the necessary files for a new pipeline:

php artisan datamorph:make product

This command will create the following files in the app/ETL/Product/ directory:

• ProductExtractor.php - For data extraction
• ProductTransformer.php - For data transformation
• ProductLoader.php - For loading transformed data

Structure of Generated Files

Extractor

<?phpdeclare(strict_types=1);
namespaceApp\ETL\Product;
useFlow\ETL\FlowContext;
usePollora\Datamorph\Contracts\Extractor;
class ProductExtractor extends Extractor
{
publicfunctionhandle(FlowContext$context): array
{
// Data extraction logic// Returns an array of raw datareturn [];
}
}

Transformer

<?phpdeclare(strict_types=1);
namespaceApp\ETL\Product;
useFlow\ETL\FlowContext;
usePollora\Datamorph\Contracts\Transformer;
class ProductTransformer extends Transformer
{
publicfunctionhandle(array$rows, FlowContext$context): array
{
// Data transformation logic// Receives raw data and returns transformed datareturn$rows;
}
}

Loader

<?phpdeclare(strict_types=1);
namespaceApp\ETL\Product;
useFlow\ETL\Rows;
useFlow\ETL\FlowContext;
usePollora\Datamorph\Contracts\Loader;
useFlow\ETL\LoaderasFlowLoader;
class ProductLoader extends Loader
{
publicfunctionhandle(FlowContext$context): FlowLoader
{
// Data loading logic// Returns a Flow ETL loaderreturnto_memory();
}
}

Running an ETL Pipeline

Once your components are implemented and your pipeline is configured, you can run it with the Artisan command:

php artisan datamorph:run stock

This command:

1. Checks that the pipeline exists in the configuration
2. Checks that the Extractor, Transformer, and Loader classes exist
3. Instantiates these classes and creates a Pipeline
4. Runs the Pipeline with the configured hooks

Running in Code

You can also run a pipeline programmatically:

usePollora\Datamorph\Pipeline;
useApp\ETL\Stock\StockExtractor;
useApp\ETL\Stock\StockTransformer;
useApp\ETL\Stock\StockLoader;
$pipeline = newPipeline(
'stock',
newStockExtractor(),
newStockTransformer(),
newStockLoader()
);
$pipeline->run();

Hooks

Hooks are a powerful mechanism in Datamorph that allows you to intervene at different stages of an ETL pipeline. There are three ways to implement hooks in Datamorph, each with its own use cases.

Configuration-based Hooks

The first approach is to define hooks in the config/datamorph.php configuration file. This method is ideal for recurring hooks that need to be applied to every pipeline execution.

Configuration

// config/datamorph.phpreturn [
'pipelines' => [
'stock' => [
'hooks' => [
'before_extract' => [
App\ETL\Stock\Hooks\BeforeStockExtract::class,
],
'after_extract' => [
App\ETL\Stock\Hooks\AfterStockExtract::class,
],
'before_transform' => [
App\ETL\Stock\Hooks\BeforeStockTransform::class,
],
'after_transform' => [
App\ETL\Stock\Hooks\AfterStockTransform::class,
],
'before_load' => [
App\ETL\Stock\Hooks\BeforeStockLoad::class,
],
'after_load' => [
App\ETL\Stock\Hooks\AfterStockLoad::class,
],
'before_run' => [
App\ETL\Stock\Hooks\BeforeStockRun::class,
],
'after_run' => [
App\ETL\Stock\Hooks\AfterStockRun::class,
],
],
],
],
];

Hook Implementation

Each hook must implement the HookInterface:

<?phpnamespaceApp\ETL\Stock\Hooks;
useClosure;
useFlow\ETL\DataFrame;
useFlow\ETL\Filesystem\SaveMode;
usePollora\Datamorph\Contracts\HookInterface;
class BeforeStockRun implements HookInterface
{
/** * Execute the hook with the given dataframe. * * @param mixed $dataframe The dataframe to process * @param Closure|null $next The next hook to execute * @return mixed */publicfunctionhandle(mixed$dataframe, ?Closure$next = null): mixed
{
// Apply hook logicif ($dataframeinstanceof DataFrame) {
$dataframe = $dataframe->mode(SaveMode::Overwrite);
}
// Pass to the next hook in the chainreturn$next ? $next($dataframe) : $dataframe;
}
}

Dynamic Hooks

The second approach uses the $context->pipeline->after() methods to register hooks dynamically during pipeline execution. This method is particularly useful for conditional behaviors or hooks that depend on the current state of the pipeline.

Usage

// In an ETL component (Extractor, Transformer, Loader)publicfunctionhandle(FlowContext$context): array
{ // Add a hook after the current operation$context->pipeline->after(function ($dataframe) {
// Hook logic
log::info("After extraction");
return$dataframe;
}); // The operation is automatically detected// ...
}

Supported Hook Types

You can pass different types of hooks to the after() method:

1. A Closure (anonymous function) - Will be automatically wrapped in a DynamicHook
2. An instance of a class implementing HookInterface - Will be used directly
3. A class name - The class will be resolved via Laravel's IoC container

Automatic Operation Detection

If you don't explicitly specify the operation, Datamorph will detect it automatically based on the calling context:

• In an Extractor, the operation will be extract
• In a Transformer, the operation will be transform
• In a Loader, the operation will be load

Hooks via before/after Methods

The third approach is to directly implement the before() and after() methods in your Extractor, Transformer, and Loader classes. This method is the simplest and most direct for standard behaviors.

Implementation

<?phpnamespaceApp\ETL\Stock;
useFlow\ETL\FlowContext;
useIlluminate\Support\Facades\Log;
usePollora\Datamorph\Contracts\Extractor;
class StockExtractor extends Extractor
{
/** * Extract stock data. */publicfunctionhandle(FlowContext$context): array
{
// Extraction logicreturn$results;
}
/** * Method executed before extraction. */publicfunctionbefore(mixed$dataframe, FlowContext$context): mixed
{
Log::info("Preparing extraction");
return$dataframe;
}
/** * Method executed after extraction. */publicfunctionafter(mixed$dataframe, FlowContext$context): mixed
{
Log::info("Extraction completed");
return$dataframe;
}
}

The before() and after() methods are automatically called by the pipeline at the appropriate times, without any additional configuration.

Execution Order and Combining Approaches

All three approaches can be combined in the same pipeline. The execution order is as follows:

1. Hooks configured in config/datamorph.php
2. The before() method of the relevant ETL component
3. Main operation (extraction, transformation, loading)
4. The after() method of the relevant ETL component
5. Dynamic hooks registered via $pipeline->before() and $pipeline->after()

This combination offers great flexibility and can address a variety of use cases.

Usage Examples

Example 1: Configuration for Validation Before Extraction

// config/datamorph.php'hooks' => [
'before_extract' => [
App\ETL\Stock\Hooks\ValidateSourceHook::class,
],
]

// App\ETL\Stock\Hooks\ValidateSourceHook.phppublicfunctionhandle(mixed$dataframe, ?Closure$next = null): mixed
{
// Check if the data source is availableif (!$this->isSourceAvailable()) {
thrownew \RuntimeException("Data source is not available");
}
return$next($dataframe);
}

Example 2: Dynamic Hooks for Conditional Logging

// In StockExtractorpublicfunctionhandle(FlowContext$context): array
{
// Add logging hooks if in debug modeif (config('app.debug')) {
$context->pipeline->before(function ($dataframe) {
Log::debug("Starting transformation");
return$dataframe;
}, 'transform');
$context->pipeline->after(function ($dataframe) {
Log::debug("Transformation completed");
return$dataframe;
}, 'transform');
}
// ...
}

Example 3: Before/After Methods for Connection Management

// In DatabaseExtractorpublicfunctionbefore(mixed$dataframe, FlowContext$context): mixed
{
// Open database connection$this->connection = DB::connection('source');
Log::info("Database connection established");
return$dataframe;
}
publicfunctionafter(mixed$dataframe, FlowContext$context): mixed
{
// Close connection after extractionif ($this->connection) {
$this->connection = null;
Log::info("Database connection closed");
}
return$dataframe;
}

Example 4: Combining Approaches for a Complete Pipeline

// config/datamorph.php - Global hooks'hooks' => [
'before_run' => [
App\ETL\Global\Hooks\LogStartHook::class,
],
'after_run' => [
App\ETL\Global\Hooks\LogEndHook::class,
],
]
// StockExtractor.php - Before/After methods
public functionbefore(mixed $dataframe, FlowContext $context): mixed
{
// Extraction-specific preparationreturn$dataframe;
}
// In a component's handle method - Dynamic hooks
public functionhandle(FlowContext $context): array
{
// Dynamic hook for a specific caseif ($someCondition) {
$context->pipeline->after(function ($dataframe) {
// Conditional logicreturn$dataframe;
});
}
// ...
}

This combination of approaches provides you with a flexible and powerful hook system capable of addressing a variety of needs in your ETL pipelines.