Run custom Python scripts with the Custom action#

#Objective

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

Using our Software Development Kit (SDK) 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..)

#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.

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.

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). 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

#Use the helper panel

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 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.

#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)

#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":

#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.

#Default list of dependencies

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

#Examples of sample scripts

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

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

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))

#Go further

If you need training or technical assistance to implement our solutions, contact your sales representative or click on this link to get a quote and ask our Professional Services experts for a custom analysis of your project.

Ask questions, give your feedback and interact directly with the team building the Data Platform on the dedicated Discord channel.

If you need support with your OVHcloud services, create a request in our Help Centre.

Join our community of users.