# Data ManagerでのDPE SDKの使用

> この簡潔な例では、Data Managerのテーブルからデータを抽出し、データを変換した後に、そのデータを追加またはData Managerを更新する方法について説明します。

DPEにビルトインされた[集計アクション](/jp/product/dpe/actions/aggregate/index)で不十分な場合は、Pythonコードで独自のデータ変換を実行できます。 

用意された簡単な方法で、Data Managerのテーブルからデータを取得し、Data Managerのテーブルにデータを追加し直すことができます。  
以下のサンプルは、[カスタムアクション](/jp/product/dpe/actions/custom/index)のコンテキストで記述されています。必要に応じて調整してください。

```python
import sys
import pandas as pd
from logging import getLogger

from forepaas.worker.connect import connect
from forepaas.worker.connector import bulk_insert

logger = logging.getLogger(__name__)
def extract_func(event):
    try:
        logger.notice("Begin function")

        # make connection to table
        cn = connect("dwh/data_prim/")

        # option 1 : extract data with no SQL required
        df = cn.select("ref_products")
        # option 2 : extract data with custom SQL
        df = cn.query("SELECT id, price_eur, eur_usd FROM ref_products")

        # perform your custom transform in the dataframe
        df["price_usd"] = df["price_eur"] * df["eur_usd"]
        
        # reinsert your dataframe in the destination table
        stats = bulk_insert(cn, "ref_products", df) 
        
        # show insertion statistics (if DBMS compatible) 
        logger.info(stats)

        # delete rows where category_id=1
        cn.delete("ref_products", {"category_id":1})

        # update rows set price where id=1
        cn.update("ref_products", {"price_eur":10.4}, {"id":1})

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