カスタムアクションを使用してカスタム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))

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