# Custom Action

?> This article is about actions that use the Data Platform's Python data processing engine. To use Apache Spark clusters, see [Custom PySpark action](/en/product/dpe/actions/custom-pyspark/index).

A *Custom action* allows you to execute custom Python scripts in a scalable cloud cluster environment.

Using our [Software Development Kit (SDK)](#/en/technical/sdk/dpe/index) to easily interact with the different components of the platform, *Custom actions* can be used to implement a variety of use-cases such as: 
*  Execute a manipulation algorithm or ETL job on your data warehouse
*  Execute a simple data analysis or machine learning algorithm
*  Extract data from data sources not available on the Data Platform marketplace without having to create connectors for it
*  Extract real time data (like MQTT, Kafka, etc..) 

?> Custom actions benefit from the whole Data Processing Engine's feature-set, typically the power of the [segmentation to parallelize the execution](/en/product/dpe/actions/settings/segmentation) of your algorithm or the orchestration within [workflows](/en/product/dpe/workflows/index) triggered immediately or on a scheduled basis.

* [Configure a Custom action](#configure-a-custom-action)
* [Use the helper panel](#use-the-helper-panel)
* [Manage dependencies in a Custom action](#manage-dependencies)
* [Examples of sample scripts](#examples-of-sample-scripts)

---
## Configure a Custom action

In the Data Processing Engine of your Project, go in the Actions tab and click on the **New Action** button. Choose the action type *Custom*.


![Creation screen of a custom action](picts/custom-action.png)
 
Drag and drop your *.py* script onto the "Drag and drop" section.  
Alternatively, select the **Start with a boilerplate** option to get started directly on the Platform's Python interface with example code snippets.

![Creation screen of a custom action](picts/action-creation.png)

The name of the function to be executed **must be entered manually** at the top of the screen in the Information panel (here and by default: "customfunc").

You will be able to edit your source file directly in the editing interface or drop a new file if needed. When you are developing your own custom actions you can use any functions provided in the [Software Development Kit (SDK)](#/en/technical/sdk/dpe/index). to easily interact with other components of the platform.  
To read more about all the available SDK functions check out the article below:

{Discover all SDK methods}(#/en/technical/sdk/dpe/index)

---
## Use the helper panel

![DPE Custom action helper](picts/custom-action-helper.png)

The Custom action editor includes a helper panel, so you can find guidance without leaving your script:

* **Scenarios**: ready-to-use action scripts organized by category, to copy and adapt to your use case.
* **SDK guides**: documentation for the [SDK](/en/technical/sdk/dpe/index.md) methods you can call from your script.
* **Data**: browse your project's data directly from the editor, to check names and structures while you write your code.
* **FAQ**: answers to common Custom action questions.

?> The helper content is the same live catalog that powers the [Data Platform Extension](/en/product/dpe/notebooks/data-platform-extension.md) in notebooks, so it is always up to date.

--- 
## Manage dependencies

### Setting language version

You can choose the Python version of your custom action among the following:
- Python 3.11
- Python 3.9 *(default option)*

> [Workflows](/en/product/dpe/workflows/index) cannot be executed with multiple versions at once.

?> We are regularly updating the available versions to provide you with a best-practice development framework. Your existing work is not migrated to a new version as long as its language version is still supported. 

### Installing Python packages

You might need to install specific packages not included by default. You can add them in the "Python Requirements" field respecting the format used in a basic requirements file for "pip" (Python package manager) then press "ENTER" on your keyboard.

This is what it should looks like once you pressed "ENTER":

![Creation screen of a custom action](picts/action-requirements.png)
 
?> When working with a Custom Action in an [Always-up](en/product/dpe/actions/settings/index?id=execution-modes) execution environment, updating dependencies triggers a redeployment of the environment to put the changes into effect.

### Installing packages from a Git repository

You can install Python packages directly from a GitHub or GitLab repository using the `git+` prefix in your requirements:

```
git+https://github.com/{OWNER}/{REPO}.git
```

To pin a specific version, add a tag or commit hash:

```
git+https://github.com/{OWNER}/{REPO}.git@<tag>
```

#### Auto-install the latest release

To always install the most recent published release without manually tracking version tags, use `@latest`:

```
git+https://github.com/{OWNER}/{REPO}.git@latest
```

The platform detects the `git+` prefix and `@latest` suffix, then automatically resolves and substitutes the latest release tag before installing.

?> After adding or modifying a Git dependency, click **Force Build** to reinstall. You no longer need to manually update the tag or commit hash each time a new version of your module is published — but note that the latest release is not picked up automatically at runtime, a manual **Force Build** is always required.

### Default list of dependencies

!> Data Platform blocks the minors of the versions allowing bug fixes to be installed. If you need a more recent version of a library you can override it manually by adding the same package with the new version in the "Requirements" field.

Here is the list of all the packages and their version (as you could find them in a requirements file for pip) shipped with the Data Processing Engine workers:

{Discover all default Python packages}(/#/en/product/dpe/actions/custom/default-packages.md)


---
## Examples of sample scripts

### Example of the extraction of a file followed by loading it into the default dataset

```python
import logging
import sys
from forepaas.dwh import connect
from forepaas.dwh import bulk_insert

def customfunc(event):
    logger = logging.getLogger(__name__)
    
    try:
        logger.info("Begin function")
        
        # Connect to the source connector
        connector = connect("dwh/dropbox_test/consommations.csv")
        
        # Upload raw file from the source connector
        connection_str = get_raw(connector)
        
        # Connect to the source
        source = connect(connection_str)
        
        # Connect to the destination connector
        destination = connect("dwh/default_dataset/consommations")
        
        # Extract dataframe from source and bulk insert into the destination
        for df in extract(source):
            stats, error = bulk_insert(destination, "consommations", df)
            logger.info(stats)
            logger.info(error)
        
        del connector, source, destination
        logger.info("END function")
        
    except Exception as e:
        raise Exception("err:{} L:{}".format(e, sys.exc_info()[2].tb_lineno))

```

### Example of a data transfer between two datasets

```python
import logging
import sys
from forepaas.dwh import connect
from forepaas.dwh import bulk_insert

def customfunc(event):
    logger = logging.getLogger(__name__)
    
    try:
        logger.info("Begin function")
        
        # Connection to a source datastore
        connector = connect("dwh/default_dataset/consommations")
        
        # Data extraction from the source by a SELECT
        lines = connector.select("consommations", {"filter_attribute": "2018-01-01"})
        
        del connector
        
        # Treatment of each line of the data
        for line in lines:
            line["new_insight"] = (line["factor1"] + line["factor2"] * 2) / 100
        
        # Connection to the destination datastore
        connector = connect("dwh/analytics_dataset/agr_consommations")
        
        # Bulk insert into the destination
        stats, err = bulk_insert(connector, "agr_consommations", lines)
        logger.info(stats)
        logger.info(err)
        
        del connector
        logger.info("END function")
        
    except Exception as e:
        raise Exception("err:{} L:{}".format(e, sys.exc_info()[2].tb_lineno))

```