Dataflow ML について

Dataflow ML のスケーリング データ処理機能は、予測と推論のパイプラインやトレーニング用のデータの準備に使用できます。

Dataflow ML ワークフローの図。

図 1. 完全な Dataflow ML ワークフロー。

要件と制限事項

  • Dataflow ML はバッチ パイプラインとストリーミング パイプラインをサポートしています。
  • RunInference API は、Apache Beam 2.40.0 以降のバージョンでサポートされています。
  • MLTransform API は、Apache Beam 2.53.0 以降のバージョンでサポートされています。
  • モデルハンドラは、PyTorch、scikit-learn、TensorFlow、ONNX、TensorRT で使用できます。サポートされていないフレームワークには、カスタムモデル ハンドラを使用できます。

トレーニング用データの準備

予測と推論のパイプライン

Dataflow ML は、Dataflow と Apache Beam の RunInference API を組み合わせて使用します。RunInference API では、モデルの特性とプロパティを定義し、その構成を RunInference 変換に渡します。この機能を使用すると、モデルの実装の詳細を知らなくても、Dataflow パイプライン内でモデルを実行できます。TensorFlow や PyTorch など、使用するデータに最適なフレームワークを選択できます。

パイプラインで複数のモデルを実行する

RunInference 変換を使用して、複数の推論モデルを Dataflow パイプラインに追加します。コードの詳細などについて詳しくは、Apache Beam ドキュメントのマルチモデル パイプラインをご覧ください。

言語間のパイプラインを構築する

Java パイプラインで RunInference を使用するには、言語間の Python 変換を作成します。このパイプラインは、前処理、後処理、推論を行う変換を呼び出します。

詳細な手順とサンプル パイプラインについては、Java SDK から RunInference を使用するをご覧ください。

Dataflow で GPU を使用する

アクセラレータを使用するバッチ パイプラインまたはストリーミング パイプラインでは、NVIDIA GPU デバイスで Dataflow パイプラインを実行できます。詳細については、GPU で Dataflow パイプラインを実行するをご覧ください。

Dataflow ML のトラブルシューティング

このセクションでは、Dataflow ML を使用する際に役立つトラブルシューティング方法とリンクについて説明します。

スタックで各テンソルが同じサイズと想定されている

RunInference API の使用時に、さまざまなサイズの画像や長さの異なる単語のエンベディングを指定すると、次のエラーが発生する可能性があります。

File "/beam/sdks/python/apache_beam/ml/inference/pytorch_inference.py", line 232, in run_inference batched_tensors = torch.stack(key_to_tensor_list[key]) RuntimeError: stack expects each tensor to be equal size, but got [12] at entry 0 and [10] at entry 1 [while running 'PyTorchRunInference/ParDo(_RunInferenceDoFn)']

このエラーは、RunInference API がさまざまなサイズのテンソル要素をバッチ処理できないために発生します。回避策については、Apache Beam ドキュメントのテンソル要素をバッチ処理できないをご覧ください。

大規模なモデルでメモリ不足エラーを回避する

中規模または大規模の ML モデルを読み込むと、マシンのメモリが不足する可能性があります。Dataflow には、ML モデルの読み込み時にメモリ不足（OOM）エラーを回避するためのツールが用意されています。次の表を使用して、シナリオに適したアプローチを決定します。

シナリオ 解決策
モデルはメモリに収まる程度の小さなサイズです。 追加の構成なしで RunInference 変換を使用します。RunInference 変換は、スレッド間でモデルを共有します。マシンの CPU コアごとに 1 つのモデルを配置できる場合は、パイプラインでデフォルト構成を使用できます。
トレーニング方法が異なる複数のモデルが同じタスクを実行している。 モデルごとのキーを使用する。詳細については、トレーニングが異なる複数のモデルを使用して ML 推論を実行するをご覧ください。
1 つのモデルがメモリに読み込まれ、すべてのプロセスがこのモデルを共有します。

large_model パラメータを使用します。詳細については、トレーニングが異なる複数のモデルを使用して ML 推論を実行するをご覧ください。

カスタムモデル ハンドラを構築する場合は、large_model パラメータを使用する代わりに、share_model_across_processes パラメータをオーバーライドします。
マシンに読み込まれるモデルの正確な数を構成する必要があります。

読み込まれるモデルの数を正確に制御するには、model_copies パラメータを使用します。

カスタム モデルハンドラを構築する場合は、model_copies パラメータをオーバーライドします。

Dataflow のメモリ管理の詳細については、Dataflow のメモリ不足エラーのトラブルシューティングをご覧ください。

次のステップ