MaxCompute PyFG (Python Feature Generation) ジョブは、オフラインバッチで複雑な特徴量を生成します。list、map、浮動小数点、int などの複雑な ODPS 2.0 データ型をサポートします。このジョブでは、設定ファイルとコマンドラインパラメーターを使用して、生成された特徴量をビニング処理するかどうかを決定します。
方法 1： 汎用リソースグループイメージの使用
DataWorks コンソールで、[スケジューリング設定] > [リソースプロパティ] に移動します。 汎用リソースグループを選択し、最新の dataworks_pairec_task_pod イメージを選択します。
注：dataworks_pairec_task_pod イメージのリリースは pyfg の更新に遅れる場合があるため、イメージに最新の pyfg パッケージが含まれていない可能性があります。必要なバージョンを確認するには、推奨ソリューションのカスタマイズ - 機能設定で生成されたスクリプトをご参照ください。最新バージョンの pyfg を使用するには、方法 3 を使用して、リソースグループ用のカスタムイメージを作成してください。
方法 2: 依存関係のインストール (DataWorks の旧バージョン)
DataWorks コンソールにログインし、スケジューリング用の専用リソースグループを作成し、O&M アシスタントを使用して pyfg パッケージをインストールします。
DataWorks 専用リソースグループに pyfg パッケージをインストールするには、DataWorks->管理センター->リソースグループリスト->O&Mアシスタント に移動し、次のコマンドを実行します。
/home/tops/bin/pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple --upgrade --force-reinstall http://tzrec.oss-cn-beijing.aliyuncs.com/third_party/pyfg105-1.0.5-cp37-cp37m-linux_x86_64.whlよくある質問
方法 3: リソースグループイメージのカスタマイズ (DataWorks の新しいバージョン向け)
手順については、「カスタムイメージ」をご参照ください。
リソースファイルのアップロード
FG 設定ファイル (JSON 形式) を MaxCompute プロジェクトにアップロードします。
一部の特徴量オペレーターには、追加のリソースファイルが必要です。これらのファイルを MaxCompute プロジェクトに手動でアップロードする必要があります。
特徴量オペレーター | 説明 | リソースファイルパラメーター |
テキスト正規化 | ストップワードファイル | |
テキストトークン化特徴量 | 語彙設定ファイル | |
テキスト関連性特徴量 | 単語頻度設定ファイル | |
カスタムオペレーター | オペレーター設定ファイル |
出力テーブルの作成
DataWorks で、PyOdps3 ノードを作成し、次のスクリプトを実行します。このスクリプトは fg.json を読み取り、出力テーブルと後続の実行に必要なリソースを作成します。
from pyfg105 import run_on_odps
fg_task = run_on_odps.FgTask(
args['input_table'],
args['output_table'],
args['fg_json_file'],
args['partition_value'],
force_delete_output_table=True,
force_update_resource=True)
fg_task.create_output_table(o)スクリプトを実行する前に、スケジューリング設定で次のパラメーターを設定してください: input_table、output_table、fg_json_file、および partition_value。
fg_task.run(o) も、出力テーブルが存在しない場合に自動的に作成しますが、このメソッドを呼び出して事前に出力テーブルを作成することをお勧めします。これにより、データを並行してバックフィルする際の競合やタスクの失敗を防ぐことができます。
FG オフラインタスクの実行
DataWorks で PyOdps3 ノードを作成し、次のスクリプトを実行します。このスクリプトは特徴量生成タスクを実行し、出力テーブルが存在しない場合は自動的に作成します。
from pyfg105 import run_on_odps
fg_task = run_on_odps.FgTask(
args['input_table'],
args['output_table'],
args['fg_json_file'],
args['partition_value'],
batch_size=128,
force_delete_output_table=False,
force_update_resource=False)
fg_task.add_sql_setting('odps.stage.mapper.split.size', 256)
fg_task.run(o)
スクリプトを実行する前に、スケジューリング設定で次のパラメーターを設定します: input_table、output_table、fg_json_file、および partition_value。
PyODPS がインストールされている場合は、pyfg をインストールしてローカルでタスクを送信することもできます。
パラメーター
パラメーター | デフォルト | 説明 |
input_table | None | 入力テーブル。 |
output_table | None | 出力テーブル。存在しない場合は自動的に作成されます。 |
fg_json_file | None | FG 設定ファイル (JSON 形式)。 |
partition_value | None | 処理する入力テーブルのパーティション。結果は出力テーブルの対応するパーティションに書き込まれます。 |
schema | None | MaxCompute スキーマ。詳細については、「スキーマ操作」をご参照ください。 |
batch_size | 128 | 各バッチで処理するレコード数。 |
memory | 1024 | タスクノードに割り当てるメモリ量 (MiB 単位)。 |
force_delete_output_table | False |
|
force_update_resource | False |
|
output_merged_str | False |
|
debug | False |
|
sql_setting | None | MaxCompute SQL パラメーター。 |
fg_setting | None | FG パラメーターは、 |
例に示すように、FgTask コンストラクタにパラメーターを渡すことで、デフォルトのパラメーター値を上書きできます。
仕組み
pyfg パッケージは、専用リソースグループ内のゲートウェイマシン上で実行され、MaxCompute に SQL タスクを送信します。または、pyfg と PyODPS の両方をローカルマシンにインストールして、タスクを送信することもできます。
SQL タスク内のカスタム UDF には、FG 共有ライブラリ、設定ファイル (fg.json、辞書、カスタムオペレーターライブラリなど)、UDF コードファイル (.py) などのリソースが必要です。これらのリソースはすべて MaxCompute クラスターにアップロードし、その分散ファイルシステムに保存する必要があります。タスクが実行されると、各ワーカーは分散ファイルシステムから必要なリソースをダウンロードし、メモリにロードします。
FG 共有ライブラリや UDF コードファイルなど、一部のリソースは複数のタスクで共有されます。force_update_resource=True の場合、システムは新しいリソースをアップロードする前に元のリソースを削除します。このプロセスでは、実行中の他のタスクを中断させる可能性のあるタイムウィンドウが発生します。