Transform foreign import source

Often the import files are not ready to be imported, when we receive them from the foreign systems. Usually there data missing, which needs to filled with Magento insides like category paths or website codes.

Therefor there is a transformation step inside the given import pipelines. In this example we will replace the default executor with custom logic.

Create an own module

First of all we need to introduce a custom module. Please refer to Create a New Module in Magento’s developer documentation.

In this example we name the module MyModule_CustomCatalogImport.

mkdir -p app/code/MyModule/CustomCatalogImport/etc
touch app/code/MyModule/CustomCatalogImport/etc/module.xml
touch app/code/MyModule/CustomCatalogImport/registration.php
Content of registration.php
Content of module.xml
<?xml version="1.0"?>
<config xmlns:xsi="" xsi:noNamespaceSchemaLocation="urn:magento:framework:Module/etc/module.xsd">
    <module name="MyModule_CustomCatalogImport" setup_version="1.0.0">
            <module name="TechDivision_PacemakerImportCatalog"/>

We need to specify TechDivision_PacemakerImportCatalog module in the sequence of our new module.xml to be able to overwrite the existing pipeline configuration.

Manipulate the pipeline

By creating a own pipeline.xml in the etc folder of our module, we can create own pipelines and overwrite existing. In this example we want to change the executor of the step product_transformation step within the pacemaker_import_catalog pipeline.

Content of app/code/MyModule/CustomCatalogImport/etc/pipeline.xml
<?xml version="1.0"?>
<config xmlns:xsi="" xsi:noNamespaceSchemaLocation="urn:magento:module:TechDivision_ProcessPipelines:etc/pipeline.xsd">
    <pipeline name="pacemaker_import_catalog">
        <step name="product_transformation" executorType="MyModule\CustomCatalogImport\Model\Executor\TransformationExecutor" />

We create a pipeline node with the name "pacemaker_import_catalog" and add here a single step node with the name "product_transformation". This both nodes are merged between the pipelines. By defining an own executorType we will change the previous defined executor, since the sequence in the module.xml of our module gives our module higher priority.

Transformation logic

Once we created our module and changed the executor for the transformation step, we need to create the executor class. In the previous part we already chose the class name MyModule\CustomCatalogImport\Model\Executor\TransformationExecutor, which means we need to create the file app/code/MyModule/CustomCatalog/Import/Model/Executor/TransformationExecutor.php according to the PSR-4 autoloader.

We suggest the directory structure Model/Executor, Model/Condition/Step and Model/Condition/Pipeline for the corresponding classes.

Place following code we already have a valid executor.


namespace MyModule\CustomCatalogImport\Model\Executor;

use TechDivision\ProcessPipelines\Api\StepInterface;
use TechDivision\ProcessPipelines\Model\Executor\AbstractExecutor;

class TransformationExecutor extends AbstractExecutor
     * @inheritdoc
    public function process(StepInterface $step): void


In order to test your current implementation you can add following code into the body of your process method and execute the pipelines.

$this->getLogger()->info("It works ;)");

Use the CLI to observe the progress of your pipeline (bin/magento pipeline:status -w 2). See the details for your pipeline with bin/magento pipeline:detail <PIPELINE_ID> and retrieve the log for your transformation step with bin/magento pipeline:step:show <STEP_ID>. The result should be something like

--- ATTEMPT 1 ---
[2020-02-11 11:10:40] executor.INFO: It works ;)

Now you’re able to add your custom logic to this executor to transform the given files. The file paths for this import pipeline are present in the $step as arguments. Execute $files = (array)$step->getArgumentValueByKey('files'); to fetch a list of file paths.


Checkout following resource for an example module. This example shows how to implement a mapping layer, which transforms data from CSV to CSV before running the import.