PHP-ETL - Operations

Building Blocks - Chain Merge

The ChainMergeConfig operation executes multiple chains of operations (branches) with the same input data, then combines their results back into the main chain. Each branch can transform the data differently, and all outputs from all branches are passed to the next step.

Key characteristics:

Each branch receives the same input data
Each branch can transform data differently
All branch outputs are combined and sent to the next step
One input item can become multiple output items
Useful for creating multiple views, enriching from multiple sources, or denormalizing data

Warning: If branches don’t filter or modify items, subsequent steps will receive duplicate data, as merge doesn’t handle duplicates automatically. This behavior can be leveraged to create multiple versions of an item.

Configuration

Use ChainMergeConfig and add branches with the addMerge() method:

use Oliverde8\Component\PhpEtl\OperationConfig\ChainMergeConfig;
use Oliverde8\Component\PhpEtl\ChainConfig;

$mergeConfig = new ChainMergeConfig();
$mergeConfig
    ->addMerge($branch1ChainConfig)
    ->addMerge($branch2ChainConfig)
    ->addMerge($branch3ChainConfig);

Each branch is a ChainConfig that can contain any sequence of operations.

addMerge() takes an optional $name, used as the branch’s identifier instead of its numeric position — the same naming ChainConfig::addLink() already supports for individual links:

$mergeConfig = new ChainMergeConfig();
$mergeConfig
    ->addMerge($simpleChainConfig, 'simple')
    ->addMerge($configurableChainConfig, 'configurable');

Naming is optional and can be mixed with unnamed branches; a duplicate name throws \InvalidArgumentException.

Parameters:

isolateContext: When true, each branch runs against its own clone of the execution context instead of sharing the parent’s. Default false.

Isolating Context

By default, all branches share the same execution context as the main chain — a parameter set with $context->setParameter() inside one branch is visible in every other branch and in the main chain once the merge is done. Pass isolateContext: true to give each branch its own independent copy instead:

$mergeConfig = new ChainMergeConfig(
    // ...
    isolateContext: true
);

Branches can then no longer see each other’s context changes, and nothing leaks back to the main chain. The file system and logger stay shared — only context parameters are isolated.

See Execution Context for more.

Example: Creating Multiple Product Variants

Here’s an example that creates both simple and configurable product records from a single input:

use Oliverde8\Component\PhpEtl\ChainConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\Extract\CsvExtractConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\ChainMergeConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\Transformer\RuleTransformConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\Loader\CsvFileWriterConfig;
use Oliverde8\Component\PhpEtl\Item\DataItem;

$chainConfig = new ChainConfig();

$chainConfig
    ->addLink(new CsvExtractConfig())
    ->addLink(
        (new ChainMergeConfig())
            // Branch 1: Create simple product
            ->addMerge(
                (new ChainConfig())
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('sku', [
                            ['implode' => [
                                'values' => [
                                    [['get' => ['field' => 'sku']]],
                                    [['constant' => ['value' => '-simple']]],
                                ],
                                'with' => '',
                            ]]
                        ])
                        ->addColumn('type', [['constant' => ['value' => 'simple']]])
                        ->addColumn('name', [['get' => ['field' => 'name']]])
                    )
            )
            // Branch 2: Create configurable product
            ->addMerge(
                (new ChainConfig())
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('sku', [
                            ['implode' => [
                                'values' => [
                                    [['get' => ['field' => 'sku']]],
                                    [['constant' => ['value' => '-configurable']]],
                                ],
                                'with' => '',
                            ]]
                        ])
                        ->addColumn('type', [['constant' => ['value' => 'configurable']]])
                        ->addColumn('name', [['get' => ['field' => 'name']]])
                    )
            )
    )
    // All branch outputs (2 products per input) go to the CSV file
    ->addLink(new CsvFileWriterConfig('merged-products.csv'));

$chainProcessor = $chainBuilder->createChain($chainConfig);
$chainProcessor->process(
    new ArrayIterator([new DataItem(['file' => 'products.csv'])]),
    []
);

Result: Each input product becomes 2 output rows - one simple, one configurable.

Example: Customer Data with Multiple Views

Create different data views for different systems from the same customer data:

$chainConfig = new ChainConfig();

$chainConfig
    ->addLink(new CsvExtractConfig())
    ->addLink(
        (new ChainMergeConfig())
            // Branch 1: Contact information view
            ->addMerge(
                (new ChainConfig())
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('customer_id', [['get' => ['field' => 'ID']]])
                        ->addColumn('full_name', [
                            ['implode' => [
                                'values' => [
                                    [['get' => ['field' => 'FirstName']]],
                                    [['get' => ['field' => 'LastName']]],
                                ],
                                'with' => ' ',
                            ]]
                        ])
                        ->addColumn('email', [['get' => ['field' => 'Email']]])
                        ->addColumn('data_type', [['constant' => ['value' => 'contact']]])
                    )
            )
            // Branch 2: Subscription status view
            ->addMerge(
                (new ChainConfig())
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('customer_id', [['get' => ['field' => 'ID']]])
                        ->addColumn('is_subscribed', [['get' => ['field' => 'IsSubscribed']]])
                        ->addColumn('subscription_date', [['get' => ['field' => 'CreatedAt']]])
                        ->addColumn('data_type', [['constant' => ['value' => 'subscription']]])
                    )
            )
    )
    ->addLink(new CsvFileWriterConfig('customer-views.csv'));

Result: Each customer becomes 2 rows - one with contact info, one with subscription data.

Example: Multi-Source API Enrichment

Enrich data by calling multiple APIs and merging all responses:

use Oliverde8\Component\PhpEtl\OperationConfig\Transformer\SimpleHttpConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\FailSafeConfig;

$chainConfig = new ChainConfig();

$chainConfig
    ->addLink(new CsvExtractConfig())
    ->addLink(
        (new ChainMergeConfig())
            // Branch 1: Get customer profile
            ->addMerge(
                (new ChainConfig())
                    ->addLink(new SimpleHttpConfig(
                        url: '@"https://api.example.com/customers/"~data["customer_id"]',
                        method: 'GET',
                        responseIsJson: true,
                        responseKey: 'profile'
                    ))
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('customer_id', [['get' => ['field' => 'customer_id']]])
                        ->addColumn('name', [['get' => ['field' => 'profile', 'name']]])
                        ->addColumn('source', [['constant' => ['value' => 'profile_api']]])
                    )
            )
            // Branch 2: Get order history (with error handling)
            ->addMerge(
                (new ChainConfig())->addLink(new FailSafeConfig(
                    chainConfig: (new ChainConfig())
                        ->addLink(new SimpleHttpConfig(
                            url: '@"https://api.example.com/orders?customer="~data["customer_id"]',
                            method: 'GET',
                            responseIsJson: true,
                            responseKey: 'orders'
                        ))
                        ->addLink((new RuleTransformConfig(false))
                            ->addColumn('customer_id', [['get' => ['field' => 'customer_id']]])
                            ->addColumn('total_orders', [['get' => ['field' => 'orders', 'total']]])
                            ->addColumn('source', [['constant' => ['value' => 'orders_api']]])
                        ),
                    exceptionsToCatch: [\Exception::class],
                    nbAttempts: 3
                ))
            )
    )
    ->addLink(new CsvFileWriterConfig('customer-enriched.csv'));

Result: Each customer becomes 2 rows (or 1 if orders API fails) with data from different APIs.

Example: Denormalizing Hierarchical Data

Split hierarchical JSON data into flat records:

use Oliverde8\Component\PhpEtl\OperationConfig\Extract\JsonExtractConfig;
use Oliverde8\Component\PhpEtl\OperationConfig\Transformer\SplitItemConfig;

// Input: {"product_id": 123, "name": "T-Shirt", "variants": ["Small", "Medium", "Large"]}

$chainConfig = new ChainConfig();

$chainConfig
    ->addLink(new JsonExtractConfig())
    ->addLink(
        (new ChainMergeConfig())
            // Branch 1: Base product info
            ->addMerge(
                (new ChainConfig())
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('product_id', [['get' => ['field' => 'product_id']]])
                        ->addColumn('name', [['get' => ['field' => 'name']]])
                        ->addColumn('type', [['constant' => ['value' => 'base']]])
                        ->addColumn('variant', [['constant' => ['value' => null]]])
                    )
            )
            // Branch 2: Each variant becomes a separate row
            ->addMerge(
                (new ChainConfig())
                    ->addLink(new SplitItemConfig(keys: ['variants']))
                    ->addLink((new RuleTransformConfig(false))
                        ->addColumn('product_id', [['get' => ['field' => 'product_id']]])
                        ->addColumn('name', [['get' => ['field' => 'name']]])
                        ->addColumn('type', [['constant' => ['value' => 'variant']]])
                        ->addColumn('variant', [['get' => ['field' => 'data']]])
                    )
            )
    )
    ->addLink(new CsvFileWriterConfig('products-flat.csv'));

Result: 1 base row + 3 variant rows = 4 total rows per product.

Common Use Cases

Multi-view data export: Create different data views for different systems
Product variants: Generate simple and configurable products from base data
Data enrichment: Combine information from multiple APIs or databases
Customer 360: Merge profile, orders, support tickets into comprehensive view