Sometimes you need to handle a complex file format beyond our Load Action capabilities
Objective
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 Data Platform Python SDK, the Datastore connector is used to interact with the Data Platform Buckets. You may think of the Datastore simply as a bucket container.
The following sample is written in Custom Action context. You may adapt it as needed.
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))
Tip
This also works with any Object Store that you may define as S31 compatible source in the Connectors.
Another example
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 forepaas/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='forepaas/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')
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.
