We have prepared detailed guides and sample scripts for common use cases to help you get started quickly and efficiently
Objective
We have prepared detailed guides and sample scripts for common use cases to help you get started quickly and efficiently. These examples demonstrate how to leverage Custom Actions in different scenarios, allowing you to extract, transform, and load (ETL) data across various components of your data platform.
1. Custom Action with a Lakehouse Manager Table
Overview:
This short example shows how to extract data from a table in Lakehouse Manager, transform it and then insert it or update the Lakehouse Manager.
The code is written in Custom Action context and it uses the stations_rides table from the Getting Started Tutorial. If you did that tutorial, you can just copy and paste the code below to test it, otherwise you need to adapt it to your tables and data sources.
Example Application:
Warning
Do not forget to build the table you use in the Lakehouse Manager and then load it with a DPE Load Action. You need to load your table before using the code below, otherwise it will not work.
import sys
import pandas as pd
import logging
from forepaas.dwh import connect
from forepaas.dwh import bulk_insert
logger = logging.getLogger(__name__)
def customfunc(event):
try:
logger.notice("Begin function")
# make connection to the default dataset
cn = connect("dwh/default_dataset/")
# option 1 : extract data from the table with no SQL required
df = cn.select("stations_rides")
# option 2 : extract data with custom SQL
df = cn.query("SELECT station_id, date, rides, station_name FROM stations_rides")
# perform your custom transform in the dataframe
df.loc[df["station_name"] == 'Harlem-Lake', "rides"] = 0
# reinsert your dataframe in the destination table
stats = bulk_insert(cn, "stations_rides", df)
# show insertion statistics (if DBMS compatible)
logger.info(stats)
# delete rows where station name is "Davis"
cn.delete("stations_rides", {"station_name":"Davis"})
# update rows set rides to 0 where station_id=40040
cn.update("stations_rides", {"rides":0}, {"station_id":40040})
# when finished, disconnect cn
del cn
logger.notice("END function")
except Exception as err:
raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))
Overview:
Sometimes you need to handle a complex file format beyond our Load Action capabilities.
In this case, we advise you to store and manipulate files with the Data Platform Buckets in your Project.
Info
In the current Data Platform SDK, the Datastore connector is used to interact with the Data Platform Buckets. You may think of the Datastore simply as a bucket container.
Example Applications:
import sys
import pandas as pd
from logging import getLogger
from forepaas.dwh import connect
from forepaas.dwh import bulk_insert
logger = getLogger(__name__)
def extract_func(event):
try:
# we get data from a bucket and we will archive them in another bucket
bucket_source_name = "your_source_bucket_name_here"
bucket_archives_name = "your_source_bucket_name_here"
# create a connector to handle bucket
bucket_connector = connect("data_store/{}".format(bucket_source_name))
# list files from bucket
files = bucket_connector.list()
# retrieve a file from Data Store bucket to temporary local folder
bucket_filepath = "stations_rides.csv"
local_filepath = "/tmp/stations_rides.csv"
bucket_connector.fget(bucket_filepath, local_filepath)
# read then transform the file as you need
# here the date column format is simply adjusted for compatibility reasons
df = pd.read_csv(local_filepath, sep=';')
df['date'] = pd.to_datetime(df['date'])
# load the dataframe into a project table named 'raw_file'
cn = connect("dwh/default_dataset/")
bulk_insert(cn, "stations_rides_artur", df)
del cn
# option 1 : copy the file into the archives bucket
bucket_archive_filepath = "archives/stations_rides.csv"
bucket_connector.fcopy_to(bucket_archives_name, bucket_archive_filepath, bucket_filepath)
# option 2 : put a file into the archives
bucket_archives = connect("data_store/{}".format(bucket_archives_name))
bucket_archives.fput(bucket_archive_filepath, local_filepath)
del bucket_archives
# delete file from source bucket
bucket_connector.delete(bucket_filepath)
# disconnect from datastore
del bucket_connector
except Exception as err:
raise Exception("err:{} L:{}".format(err,sys.exc_info()[2].tb_lineno))
Below is an example code which uploads an image to a bucket from a simple URL.
from forepaas.dwh import connect
data_store = connect('data_store')
# Get bucket and upload image from URL to path uploads/test.jpg.
# And finally get the image from the bucket
bucket_test = data_store.get_bucket('test')
lists = bucket_test.list(recursive=True)
bucket_test.put_request("https://i.stack.imgur.com/r8jTK.jpg", path='uploads/test.jpg')
data = bucket.get('hello/test.jpg')
# Create a bucket if it does not already exists
if data_store.bucket_exists('test-exists') is False:
data_store.create_bucket('test-exists')
# Connect directly to the bucket test and remove the file
bucket_test2 = connect('data_store/test')
bucket_test2.delete('hello/test.jpg')
Tip
This also works with any Object Store that you may define as S31 compatible source in the Connectors.
3. Custom Action with a Connectors Source
Overview:
This example will show you how to get a file directly from a Connectors Source, treat it and put the treated data into a Lakehouse Manager Table.
The code is written in the Custom Action context and it uses the chicago_files source from the Getting Started Tutorial. If you did that tutorial, you can just copy and paste the code below to test it, otherwise you need to adapt it to your tables and data sources.
Tip
This works with any Source protocol such as FTP, Dropbox, etc.
Example Applications:
import sys
from forepaas.dwh.connect import connect
from forepaas.dwh import bulk_insert
import logging
logger = logging.getLogger(__name__)
def customfunc(event):
try:
# here we are connecting to a source named 'chicago_files'
source_address = "dwh/chicago_files_artur/"
# specify unsupported filename from list of files in source
filename_w_extension = "stations_rides.csv"
# connecto to file toget the address
source_file_connector = connect(source_address + filename_w_extension)
# connect to file directly
file_address = source_file_connector.get()
file_connector = connect(file_address)
# Extract and treat the file so it is usable
df = file_connector.extract(return_type='dataframe')
# Treat the data
# - - - -
# connect to the Lakehouse Manager
dm_connector = connect("dwh/default_dataset/")
# insert into an existing destination table
stats = bulk_insert(dm_connector, "chicago_calendar_full", df)
logger.info(stats)
# disconnect from datastore and remote source
del source_connector
del dm_connector
except Exception as err:
raise Exception(f"err:{err} L:{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.
1: S3 is a trademark of Amazon Technologies, Inc. OVHcloud's service is not sponsored by, endorsed by, or otherwise affiliated with Amazon Technologies, Inc.
