Featuretoolsフレームワークの使用経験

今日、多くの企業にとって最も重要な開発のベクトルはデジタル化です。そして、ほとんどの場合、それは何らかの形で機械学習に関連しています。つまり、機能を数える必要があるモデルに関連しています。



手動で行うこともできますが、このプロセスをより速く簡単にするためのフレームワークとライブラリもあります。今日は



、そのうちの1つであるfeaturetoolsと、それを使用した実際の経験について説明します。





最もファッショナブルなパイプライン



こんにちは！私はアレクサンダー・ロスクトフです。リロイ・マーリンでデータアナリストとして働いています。今では流行りのデータ科学者として働いています。私の責任には、データの操作、分析クエリとアンロードから始まり、モデルのトレーニング、サービスなどでのモデルのラップ、コードの配信と展開の設定、およびその作業の監視が含まれます。



元に戻す予測は、私が取り組んでいる製品の1つです。



製品の目的：顧客がオンライン注文をキャンセルする可能性を予測します。このような予測の助けを借りて、注文を行った顧客のどれを最初に呼び出して注文の確認を求めるべきか、そして誰がまったく呼び出されないかを判断できます。第一に、電話でクライアントからの注文を電話で確認するという事実自体がキャンセルの可能性を減らします。第二に、電話をかけて相手が拒否した場合、リソースを節約できます。注文の収集に費やしたであろう従業員のためにより多くの時間が解放されます。さらに、この方法で製品は棚に残り、その時点で店の顧客がそれを必要とする場合、彼はそれを購入することができます。そしてこれにより、後でキャンセルされた注文で収集され、棚になかった商品の数が減ります。

先駆者

製品パイロットについては、複数の店舗での集荷のために後払い注文のみを受け付けます。



既製のソリューションは次のように機能します。ApacheNiFiを使用して注文が届き、商品に関するデータを取得するなど、情報を充実させます。次に、これらすべてがApacheKafkaメッセージブローカーを介してPythonで記述されたサービスに転送されます。サービスは注文の特徴を計算し、次にそれらの機械学習モデルが採用され、キャンセルの確率が示されます。その後、ビジネスロジックに従って、今すぐクライアントに電話する必要があるかどうかの回答を準備します（たとえば、店内の従業員の助けを借りて注文した場合、または夜に注文した場合は、電話しないでください）。



何が連続して全員に電話をかけるのを妨げるのでしょうか？事実、電話をかけるためのリソースの数は限られているため、誰が確実に電話をかけるべきか、誰が電話なしで確実に注文を受け取るかを理解することが重要です。

モデル開発

私はサービス、モデル、それに応じてモデルの特徴の計算に従事しました。これについては後で説明します。



トレーニング中に特徴を計算するときは、3つのデータソースを使用します。

1. 注文メタ情報（注文番号、タイムスタンプ、顧客のデバイス、配送方法、支払い方法）が記載されたプレート。
2. レシートの位置が記載されたプレート：注文番号、商品、価格、数量、在庫品の数量。各ポジションは別々の行にあります。
3. 表-商品のリファレンスブック：記事、商品のカテゴリ、測定単位、説明を含むいくつかのフィールド。

標準のPythonメソッドとpandasライブラリを使用すると、すべてのテーブルを1つの大きなテーブルに簡単に組み合わせることができます。その後、groupbyを使用して、順序別の集計、製品別の履歴、製品カテゴリ別など、あらゆる種類の機能を計算できます。ただし、ここでいくつかの問題が発生します。

• 計算の並列処理。標準のgroupbyは1つのスレッドで機能し、ビッグデータ（最大1,000万行）では、残りのコアの容量がアイドル状態である間、100の機能が許容できないほど長いと見なされます。
• コードの量：このような各リクエストは個別に記述し、正確性をチェックする必要があります。その後、すべての結果を収集する必要があります。これには時間がかかります。特に、レシート内のアイテムの最新の履歴を計算し、注文のこれらの特性を集計するなど、一部の計算が複雑であることを考えると、これには時間がかかります。
• すべてを手作業でコーディングすると、間違いを犯す可能性があります。

「私たちはすべてを手で書く」アプローチの利点は、もちろん、行動の完全な自由であり、あなたはあなたの想像力を展開することができます。



疑問が生じます：作業のこの部分をどのように最適化できるか。1つの解決策は、featuretoolsライブラリを使用することです。



ここでは、この記事の本質、つまりライブラリ自体とその使用方法にすでに進んでいます。

なぜfeaturetools？

プレートの形で機械学習のさまざまなフレームワークを考えてみましょう（写真自体はここから正直に盗まれており、おそらくすべてがそこに示されているわけではありませんが、それでも）：







私たちは主に機能エンジニアリングブロックに関心があります。これらすべてのフレームワークとパッケージを見ると、featuretoolsが最も洗練されており、tsfreshなどの他のライブラリの機能も含まれていることがわかります。



また、featuretoolsの利点（広告はまったくありません！）は次のとおりです。

• 箱から出して並列コンピューティング
• 箱から出して多くの機能の可用性
• カスタマイズの柔軟性-非常に複雑なことを考慮することができます
• 異なるテーブル間の関係の説明（リレーショナル）
• より少ないコード
• 間違いを犯しにくい
• それ自体では、登録やSMSなしですべてが無料です（ただし、pypiを使用）

しかし、それはそれほど単純ではありません。

• フレームワークにはある程度の学習が必要であり、完全な習得にはかなりの時間がかかります。
• 最も人気のある質問はまだよくグーグルですが、それはそれほど大きなコミュニティを持っていません。
• 使用自体も、フィーチャスペースを不必要に膨らませたり、計算時間を長くしたりしないように注意する必要があります。

トレーニング

featuretoolsの構成例を示します。



次に、簡単な説明が記載されたコードがあります。機能ツール、そのクラス、メソッド、機能について詳しく説明されています。特に、フレームワークのWebサイトのドキュメントで読むことができます。実際のタスクでいくつかの興味深い可能性を示す実用的なアプリケーションの例に興味がある場合は、コメントを書き込んでください。おそらく別の記事を書き留めます。



そう。



最初に、EntitySetクラスのオブジェクトを作成する必要があります。このオブジェクトには、データを含むテーブルが含まれており、それらの相互関係を認識しています。



データを含む3つのテーブルがあることを思い出してください。

• orders_meta（注文メタ情報）
• orders_items_lists（注文のアイテムに関する情報）
• アイテム（記事とそのプロパティの参照）

次のように記述します（さらに、3つのストアのデータのみが使用されます）。

import featuretools as ft

es = ft.EntitySet(id='orders')  #     EntitySet

#      pandas.DataFrame-   (ft.Entity)
es = es.entity_from_dataframe(entity_id='orders_meta',
                              dataframe=orders_meta,
                              index='order_id',
                              time_index='order_creation_dt')
es = es.entity_from_dataframe(entity_id='orders_items',
                              dataframe=orders_items_lists,
                              index='order_item_id')
es = es.entity_from_dataframe(entity_id='items',
                              dataframe=items,
                              index='item',
                              variable_types={
                                  'subclass': ft.variable_types.Categorical
                              })

#    

#      -,
#  -
relationship_orders_items_list = ft.Relationship(es['orders_meta']['order_id'],
                                                 es['orders_items']['order_id'])
relationship_items_list_items = ft.Relationship(es['items']['item'],
                                                es['orders_items']['item'])

#  
es = es.add_relationship(relationship_orders_items_list)
es = es.add_relationship(relationship_items_list_items)

やったー！これで、あらゆる種類の兆候を数えることができるオブジェクトができました。



かなり単純な機能を計算するためのコードを提供します。注文ごとに、商品の価格と数量に関するさまざまな統計、時間ごとのいくつかの機能、注文内の最も頻繁な製品と商品のカテゴリを計算します（データを使用してさまざまな変換を実行する関数は、featuretoolsではプリミティブと呼ばれます） ..。

orders_aggs, orders_aggs_cols = ft.dfs(
    entityset=es,
    target_entity='orders_meta',
    agg_primitives=['mean', 'count', 'mode', 'any'],
    trans_primitives=['hour', 'weekday'],
    instance_ids=[200315133283, 200315129511, 200315130383],
    max_depth=2
)

テーブルにはブール列がないため、プリミティブは適用されませんでした。一般に、featuretools自体がデータタイプを分析し、適切な関数のみを適用すると便利です。



また、手動で計算する注文をいくつか指定しました。これにより、計算をすばやくデバッグできます（何か間違った構成をした場合はどうなりますか）。



それでは、機能にさらにいくつかの集計、つまりパーセンタイルを追加しましょう。ただし、featuretoolsには、それらを計算するための組み込みプリミティブがありません。だからあなたはそれを自分で書く必要があります。

from featuretools.variable_types import Numeric
from featuretools.primitives.base.aggregation_primitive_base import make_agg_primitive


def percentile05(x: pandas.Series) -> float:
   return numpy.percentile(x, 5)


def percentile25(x: pandas.Series) -> float:
   return numpy.percentile(x, 25)


def percentile75(x: pandas.Series) -> float:
   return numpy.percentile(x, 75)


def percentile95(x: pandas.Series) -> float:
   return numpy.percentile(x, 95)


percentiles = [percentile05, percentile25, percentile75, percentile95]
custom_agg_primitives = [make_agg_primitive(function=fun,
                                            input_types=[Numeric],
                                            return_type=Numeric,
                                            name=fun.__name__)
                         for fun in percentiles]

そして、それらを計算に追加します。

orders_aggs, orders_aggs_cols = ft.dfs(
    entityset=es,
    target_entity='orders_meta',
    agg_primitives=['mean', 'count', 'mode', 'any'] + custom_agg_primitives,
    trans_primitives=['hour', 'weekday'],
    instance_ids=[200315133283, 200315129511, 200315130383],
    max_depth=2
)

その後、すべてが同じです。これまでのところ、すべてが非常にシンプルで簡単です（もちろん比較的）。



機能計算機を保存して、モデル実行の段階、つまりサービスで使用したい場合はどうなりますか？

戦闘中のFeaturetools

ここから主な問題が始まります。



受注の特性を計算するには、EntitySetを再度作成してすべての操作を実行する必要があります。また、大きなテーブルの場合、pandas.DataFrameオブジェクトをEntitySetにスローするのはごく普通のようですが、1つの行からDataFrameに対して同様の操作を実行します（テーブルにはチェック付きのオブジェクトが多くありますが、チェックごとに平均3.3の位置、つまり十分ではありません）-それほどでもない。結局のところ、そのようなオブジェクトの作成とその助けを借りた計算には、必然的にオーバーヘッドが含まれます。つまり、たとえば、任意のサイズのオブジェクトを作成するときのメモリ割り当てと初期化、または複数の機能を同時に計算するときの並列化プロセス自体に必要な、削除できない数の操作が含まれます。



したがって、当社の製品の「一度に1つの注文」モードでは、機能ツールは最高の効率を示さず、サービス応答時間の平均75％を占めます（ハードウェアに応じて平均150〜200ミリ秒）。比較のために：既製の機能でcatboostを使用して予測を計算するには、サービス応答時間の3％、つまり10ミリ秒以下かかります。



さらに、カスタムプリミティブの使用に関連する別の落とし穴があります。事実、作成したプリミティブを含むクラスのオブジェクトはピクルスされていないため、ピクルスに単純に保存することはできません。



次に、作成したプリミティブを含むFeatureBaseオブジェクトのリストを保存できる組み込みのsave_features（）関数を使用してみませんか？

それらは保存されますが、事前に再作成しないと、後でload_features（）関数を使用して読み取ることはできません。つまり、理論的にはディスクから読み取る必要のあるプリミティブを最初に作成し直して、二度と使用しないようにします。



次のようになります。

from __future__ import annotations

import multiprocessing
import pickle
from typing import List, Optional, Any, Dict

import pandas
from featuretools import EntitySet, dfs, calculate_feature_matrix, save_features, load_features
from featuretools.feature_base.feature_base import FeatureBase, AggregationFeature
from featuretools.primitives.base.aggregation_primitive_base import make_agg_primitive

# - 

#      ,
#       
#     
#        (    ),
#      ,     
class AggregationFeaturesCalculator:
    def __init__(self,
                 target_entity: str,
                 agg_primitives: List[str],
                 custom_primitives_params: Optional[List[Dict[str, Any]]] = None,
                 max_depth: int = 2,
                 drop_contains: Optional[List[str]] = None):
        if custom_primitives_params is None:
            custom_primitives_params = []
        if drop_contains is None:
            drop_contains = []
        self._target_entity = target_entity
        self._agg_primitives = agg_primitives
        self._custom_primitives_params = custom_primitives_params
        self._max_depth = max_depth
        self._drop_contains = drop_contains
        self._features = None  #   (  ft.FeatureBase)

    @property
    def features_are_built(self) -> bool:
        return self._features is not None

    @property
    def features(self) -> List[AggregationFeature]:
        if self._features is None:
            raise AttributeError('features have not been built yet')
        return self._features

    #        
    def build_features(self, entity_set: EntitySet) -> None:
        custom_primitives = [make_agg_primitive(**primitive_params)
                             for primitive_params in self._custom_primitives_params]
        self._features = dfs(
            entityset=entity_set,
            target_entity=self._target_entity,
            features_only=True,
            agg_primitives=self._agg_primitives + custom_primitives,
            trans_primitives=[],
            drop_contains=self._drop_contains,
            max_depth=self._max_depth,
            verbose=False
        )

    # ,    
    #       
    @staticmethod
    def calculate_from_features(features: List[FeatureBase],
                                entity_set: EntitySet,
                                parallelize: bool = False) -> pandas.DataFrame:
        n_jobs = max(1, multiprocessing.cpu_count() - 1) if parallelize else 1
        return calculate_feature_matrix(features=features, entityset=entity_set, n_jobs=n_jobs)

    #       
    def calculate(self, entity_set: EntitySet, parallelize: bool = False) -> pandas.DataFrame:
        if not self.features_are_built:
            self.build_features(entity_set)
        return self.calculate_from_features(features=self.features,
                                            entity_set=entity_set,
                                            parallelize=parallelize)

    #    
    
    #     ,
    #     save_features()
    #       
    @staticmethod
    def save(calculator: AggregationFeaturesCalculator, path: str) -> None:
        result = {
            'target_entity': calculator._target_entity,
            'agg_primitives': calculator._agg_primitives,
            'custom_primitives_params': calculator._custom_primitives_params,
            'max_depth': calculator._max_depth,
            'drop_contains': calculator._drop_contains
        }
        if calculator.features_are_built:
            result['features'] = save_features(calculator.features)
        with open(path, 'wb') as f:
            pickle.dump(result, f)

    #      
    @staticmethod
    def load(path: str) -> AggregationFeaturesCalculator:
        with open(path, 'rb') as f:
            arguments_dict = pickle.load(f)
        
        #    ...
        if arguments_dict['custom_primitives_params']:
            custom_primitives = [make_agg_primitive(**custom_primitive_params)
                                 for custom_primitive_params in arguments_dict['custom_primitives_params']]
        features = None
        
        #       
        if 'features' in arguments_dict:
            features = load_features(arguments_dict.pop('features'))
        calculator = AggregationFeaturesCalculator(**arguments_dict)
        if features:
            calculator._features = features
        return calculator

load（）関数では、使用されないプリミティブ（custom_primitives変数を宣言）を作成する必要があります。ただし、これがないと、load_features（）関数呼び出しの場所での機能のロードが失敗し、RuntimeError：モジュール "featuretools.primitives.base.aggregation_primitive_base"のプリミティブ "percentile05"が見つかりません。



あまり論理的ではないことがわかりますが、機能します。特定のデータ形式に既に関連付けられている計算機（機能は、値自体がなくても、計算されたEntitySetに関連付けられているため）と、指定されたプリミティブのリストのみを含む計算機の両方を保存できます。



おそらく将来的にはこれが修正され、FeatureBaseオブジェクトの任意のセットを便利に保存できるようになるでしょう。

なぜそれを使用するのですか？

開発時間の観点からは安価でありながら、既存の負荷での応答時間は余裕を持ってSLA（5秒）に収まるためです。



ただし、頻繁に受信する要求に迅速に応答する必要があるサービスの場合、非同期呼び出しのような追加の「スクワット」なしで機能ツールを使用すると問題が発生することに注意してください。



これは、学習段階と推論段階で機能ツールを使用した経験です。



このフレームワークは、トレーニング用の多数の機能をすばやく計算するためのツールとして非常に優れており、開発時間を大幅に短縮し、エラーの可能性を減らします。

撤退の段階でそれを使用するかどうかはあなたの仕事に依存します。