PAI-TensorFlowを使用して分散DeepFMアルゴリズムを実装する方法 - Platform For AI

このトピックでは、PAI-TensorFlowを使用して分散DeepFMアルゴリズムを実装する方法について説明します。

警告

GPUアクセラレーションサーバーは段階的に廃止されます。 CPUサーバーで実行されるTensorFlowタスクを送信できます。モデルトレーニングにGPU高速化インスタンスを使用する場合は、Deep Learning Containers (DLC) に移動してジョブを送信します。詳細については、「トレーニングジョブの送信」をご参照ください。

前提条件

Object Storage Service (OSS) が有効化され、バケットが作成されます。詳細については、「OSSの有効化」および「バケットの作成」をご参照ください。
重要
バケットの作成時にバージョン管理を有効にしないでください。そうしないと、同じ名前の別のオブジェクトでオブジェクトを上書きすることはできません。
Platform for AI (PAI) は、OSSバケットへのアクセスを許可されています。詳細については、「Machine Learning Designerの使用に必要な権限の付与」をご参照ください。

背景情報

DeepFMアルゴリズムは、Wide&deepアルゴリズムと同じ深い部分を持っています。ただし、DeepFMアルゴリズムでは、ロジスティック回帰 (LR) の代わりに因数分解マシン (FM) を使用して、手動機能エンジニアリングを回避します。

トレーニングデータソースはpai_online_project.dwd_avazu_ctr_deepmodel_trainです。テストデータソースはpai_online_project.dwd_avazu_ctr_deepmodel_testです。 pai_online_project.dwd_avazu_ctr_deepmodel_trainとpai_online_project.dwd_avazu_ctr_deepmodel_testは、使用できる公開データソースです。

手順

モデルファイルをダウンロードします。

モデル設定コードを変更します。

embedding_dim,hash_bucketsとdefault_value各フィーチャーのパラメーター。
```
self.fields_config_dict['hour'] = {'field_name': 'field1', 'embedding_dim': self.embedding_dim, 'hash_bucket': 50, 'default_value': '0'}
self.fields_config_dict['c1'] = {'field_name': 'field2', 'embedding_dim': self.embedding_dim, 'hash_bucket': 10, 'default_value': '0'}
```
DeepFMモデルでは、すべてのフィーチャのembeding_dimパラメーターを同じ値に設定する必要があります。この制限は、Wide&Deepモデルには適用されません。 hash_bucketsパラメーターは、user_idおよびitemidフィーチャーでは大きい値に設定し、値が少ないフィーチャーでは小さい値に設定することを推奨します。
モデルを構成します。モデルパラメーターでDeepFMまたはwdlを指定して、deepfmまたはWide&Deepモデルを設定することを推奨します。
```
tf.app.flags.DEFINE_string("model", 'deepfm', "model {'wdl', 'deepfm'}")
```
分散パラメーターを設定します。
```
tf.app.flags.DEFINE_string("job_name", "", "job name")
tf.app.flags.DEFINE_integer("task_index", None, "Worker or server index")
tf.app.flags.DEFINE_string("ps_hosts", "", "ps hosts")
tf.app.flags.DEFINE_string("worker_hosts", "", "worker hosts")
```
トレーニングタスクを送信するときは、clusterパラメーターのみを設定する必要があります。システムは自動的に分散パラメータを生成します。クラスターパラメーターの詳細については、「手順4」をご参照ください。

入力データを指定します。

  def _parse_batch_for_tabledataset(self, *args):
    label = tf.reshape(args[0], [-1])
    fields = [tf.reshape(v, [-1]) for v in args[1:]]
    return dict(zip(self.feas_name, fields)), label

  def train_input_fn_from_odps(self, data_path, epoch=10, batch_size=1024, slice_id=0, slice_count=1):
    with tf.device('/cpu:0'):
      dataset = tf.data.TableRecordDataset([data_path], record_defaults=self.record_defaults,
                         slice_count=slice_count, slice_id=slice_id)
      dataset = dataset.batch(batch_size).repeat(epoch)
      dataset = dataset.map(self._parse_batch_for_tabledataset, num_parallel_calls=8).prefetch(100)
      return dataset

  def val_input_fn_from_odps(self, data_path, epoch=1, batch_size=1024, slice_id=0, slice_count=1):
    with tf.device('/cpu:0'):
      dataset = tf.data.TableRecordDataset([data_path], record_defaults=self.record_defaults,
                         slice_count=slice_count, slice_id=slice_id)
      dataset = dataset.batch(batch_size).repeat(epoch)
      dataset = dataset.map(self._parse_batch_for_tabledataset, num_parallel_calls=8).prefetch(100)
      return dataset

フィーチャ変換が必要な場合は、MaxComputeを使用してモデル外のフィーチャを変換することを推奨します。これにより、トレーニングのオーバーヘッドを減らすことができます。モデル内でフィーチャを変換する場合は、_parse_batch_for_tabledataset関数を使用してフィーチャを変換することを推奨します。

変更したモデルファイルをOSSにアップロードします。
オプションです。 トレーニングタスクを送信します。
説明
モデルファイルをトレーニングに使用しない場合は、この手順を実行する必要があります。それ以外の場合は、この手順をスキップします。
設定されたモデルのタイプに基づいて、次のいずれかのコマンドを実行して、トレーニングタスクを送信します。
- DeepFM
```
pai -name tensorflow1120_ext
-project algo_public
-Dbuckets='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/'
-Darn=''
-Dscript='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/demo/deepfm_pai_ctr.py'
-Dtables='odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_train,odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_test'
-DuserDefinedParameters="--task_type='train' --model='deepfm' --checkpoint_dir='oss://bucket_name/path/' --output_dir='oss://bucket_name/path/'"
-Dcluster='{\"ps\":{\"count\":2,\"cpu\":1200,\"memory\":10000},\"worker\":{\"count\":8,\"cpu\":1200,\"memory\":30000}}';
```
  projectパラメーターとtablesパラメーターのpai_online_project変数は、入力データを格納するプロジェクトの名前を指定します。バケットとスクリプトパラメーターのbucket_nameパラメーターとregionパラメーターは、OSSバケットの名前とOSSバケットが存在するリージョンを別々に指定します。 arnパラメーターには、Alibaba CloudアカウントのResource Access Management (RAM) ロールのAlibaba Cloud Resource Name (ARN) を指定します。ビジネス要件に基づいてパラメーターと変数を設定できます。 ARNの詳細については、「RAMロールの概要」をご参照ください。
- ワイド&ディープ
```
pai -name tensorflow1120_ext
-project algo_public
-Dbuckets='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/'
-Darn=''
-Dscript='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/demo/deepfm_pai_ctr.py'
-Dtables='odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_train,odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_test'
-DuserDefinedParameters="--task_type='train' --model='wdl' --checkpoint_dir='oss://bucket_name/path/' --output_dir='oss://bucket_name/path/'"
-Dcluster='{\"ps\":{\"count\":2,\"cpu\":1200,\"memory\":10000},\"worker\":{\"count\":8,\"cpu\":1200,\"memory\":30000}}';
```
  projectパラメーターとtablesパラメーターのpai_online_project変数は、入力データを格納するプロジェクトの名前を指定します。バケットとスクリプトパラメーターのbucket_name変数とregion変数は、OSSバケットの名前とOSSバケットが存在するリージョンを別々に指定します。 arnパラメーターには、Alibaba CloudアカウントのRAMロールのAlibaba Cloud ARNを指定します。ビジネス要件に基づいてパラメーターと変数を設定できます。 ARNの詳細については、「RAMロールの概要」をご参照ください。
トレーニングタスクは、パラメータサーバー (PS)-ワーカーアーキテクチャで実行される分散タスクです。したがって、clusterパラメーターを設定する必要があります。上記のコードでは、2つのPSノードと8つのワーカーノードがclusterパラメーターで指定されています。各PSノードには12 CPUコアと10 GBのメモリがあります。各ワーカーノードには12個のCPUコアと30 GBのメモリがあります。上記のコードのパラメーターの詳細については、「PAI-TensorFlowタスクのパラメーター」をご参照ください。
オフライン推論タスクを送信します。
```
pai -name tensorflow1120_ext
-project algo_public
-Dbuckets='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/'
-Darn=''
-Dscript='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/demo/deepfm_pai_ctr.py'
-Dtables='odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_train,odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_test'
-DuserDefinedParameters="--task_type='predict' --model='deepfm' --checkpoint_dir='oss://bucket_name/path/' --output_dir='oss://bucket_name/path/'"
-Dcluster='{\"worker\":{\"count\":8,\"cpu\":1200,\"memory\":30000}}'
-Doutputs='odps://project_name/tables/output_table_name';
```
projectパラメーターとtablesパラメーターのpai_online_project変数は、入力データを格納するプロジェクトの名前を指定します。バケットとスクリプトパラメーターのbucket_name変数とregion変数は、OSSバケットの名前とOSSバケットが存在するリージョンを別々に指定します。 arnパラメーターには、Alibaba CloudアカウントのRAMロールのARNを指定します。ビジネス要件に基づいてパラメーターと変数を設定できます。 ARNの詳細については、「RAMロールの概要」をご参照ください。
オフライン推論タスクを送信する前に、出力テーブルを作成する必要があります。各推論タスクの結果がテーブルに書き込まれ、前の推論タスクの結果が上書きされます。次のコマンドを実行して、出力テーブルを作成します。
```
drop table project_name.output_table_name;
create table project_name.output_table_name
(
   probabilities STRING
   ,logits STRING
)STORED AS ALIORC;
```
トレーニング済みモデルファイルをエクスポートします。
```
pai -name tensorflow1120_ext
-project algo_public
-Dbuckets='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/'
-Darn=''
-Dscript='oss://bucket_name.oss-cn-region-internal.aliyuncs.com/demo/deepfm_pai_ctr.py'
-Dtables='odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_train,odps://pai_online_project/tables/dwd_avazu_ctr_predict_deepmodel_test'
-DuserDefinedParameters="--task_type='savemodel' --model='deepfm' --checkpoint_dir='oss://bucket_name/path/' --output_dir='oss://bucket_name/path/'"
-Dcluster='{\"ps\":{\"count\":2,\"cpu\":1200,\"memory\":10000},\"worker\":{\"count\":8,\"cpu\":1200,\"memory\":30000}}';
```
projectパラメーターとtablesパラメーターのpai_online_project変数は、入力データを格納するプロジェクトの名前を指定します。バケットおよびスクリプトパラメーターのbucket_name変数とregion変数は、OSSバケットの名前とOSSバケットが存在するリージョンを別々に指定します。 arnパラメーターには、Alibaba CloudアカウントのRAMロールのARNを指定します。ビジネス要件に基づいてパラメーターと変数を設定できます。 ARNの詳細については、「RAMロールの概要」をご参照ください。ワーカーノードを使用してモデルファイルをエクスポートします。