Use segmentation (and perimeter) in a Custom action

In a Custom action, you can fully benefit from the segmentation system and even go further

Objective

In a Custom action, you can fully benefit from the segmentation system and even go further:

parallelizing the calculation of your complex algorithm
parallelizing huge custom data loading
etc..

You can set the segmentation type and values as you would do in other compatible action types, through the interface, or through advanced JSON mode. For more information, please refer to the segmentation documentation page

Here, you will learn how to get the segmentation configuration and values in your custom Python script, in order to apply it as you want.
Let's dig into an example where you set a segmentation over a set of values, and apply a filter on the source.

import sys
from logging import getLogger

from forepaas.core.settings import PARAMS
from forepaas.dwh import connect
from forepaas.dwh import bulk_insert

logger = logging.getLogger(__name__)

def my_custom(event):
	try:
		# get the segmentation and perimeter settings
		segmentation = PARAMS.get("segmentation",{})
		segmentationValues = PARAMS.get("segmentation",{}).get("values")
		perimeter = PARAMS.get("perimeter",{})
		perimeterValues = PARAMS.get("perimeter",{}).get("values")

		# set a filter dict regarding the segmentation and perimeter configuration
		filters = {}
		if segmentation["type"] in ["workflow_dates", "dwh_attributes", "predefined_set"] and segmentationValues is not None: 
			filters[segmentation["var_name"]] = segmentationValues
		if perimeter["type"] in ["workflow_dates", "dwh_attributes", "predefined_set"] and perimeterValues is not None: 
			filters[perimeter["var_name"]] = perimeterValues

		cn = connect("dwh/default_dataset/")

		# extract with filters
		df = cn.select(table_name, filters)

		# perform some changes on the df optionaly 
		# df = df

		# insert into destination
		bulk_insert(cn, table_destination, df)

		# close connection
		del cn
	except Exception as err: 
		raise Exception("err:{} L:{}".format(err,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.

Esta página foi útil?