Cloud Dataflow パイプラインでの予測のための機械学習モデルの比較

Last reviewed 2018-09-07 UTC

このソリューションでは、Dataflow パイプラインで機械学習モデルを呼び出すためのさまざまな設計アプローチを説明して比較し、それぞれのアプローチを選択する際のトレードオフを検証します。バッチ処理パイプラインとストリーム処理パイプラインの両方で、さまざまなアプローチを検討し、これらのトレードオフを説明する一連のテストの結果を紹介します。このソリューションは、機械学習モデルを構築するデータサイエンティストではなく、トレーニングされたモデルをデータ処理パイプラインに統合する担当者を念頭に置いて設計されています。

概要

この ML モデルを Dataflow パイプラインに統合する担当者は、さまざまなアプローチの内容や、システム要件に最適なアプローチを知りたいと思うでしょう。以下を検討する必要があります。

スループット
レイテンシ
料金
実装
メンテナンス

これらの考慮事項のバランスをとることは必ずしも容易ではありませんが、このソリューションは、優先順位に基づいて意思決定プロセスをナビゲートするのに役立ちます。このソリューションでは、バッチデータパイプラインとストリームデータパイプラインの TensorFlow でトレーニングされた機械学習（ML）モデルで予測を行うための 3 つのアプローチを比較します。

ストリーミングパイプラインに REST / HTTP API としてデプロイされたモデルを使用する。
バッチパイプラインに AI Platform（AI Platform）バッチ予測ジョブを使用する。
バッチパイプラインとストリーミングパイプラインの両方に Dataflow 直接モデル予測を使用する。

すべてのテストで、Natality データセットと呼ばれる既存のトレーニングされたモデルを使用しています。これは、さまざまな入力に基づいて赤ちゃんの体重を予測します。このソリューションの目標はモデルを構築することではないため、モデルの構築方法やトレーニング方法については説明しません。Natality データセットの詳細については、次のステップをご覧ください。

プラットフォーム

データパイプラインを実行し、トレーニングされた ML モデルを呼び出すには、さまざまな方法があります。ただし、機能要件は常に同じです。

制限付き（バッチ）ソースまたは無制限（ストリーミング）ソースからのデータの取り込み。データを取り込むソースの例には、センサーデータ、ウェブサイトの相互作用、金融トランザクションなどがあります。
予測のために ML モデルを呼び出すことによって入力データを変換し、拡充する。たとえば、JSON ファイルを解析して関連するフィールドを抽出し、メンテナンス日の予測、プロダクトの推奨事項の作成、不正行為の検出を行います。
分析やバックアップのために変換されたデータと予測を保存するか、新しいイベントや追加のパイプラインをトリガーするキューシステムに渡す。たとえば、潜在的な不正行為をリアルタイムで検出する方法や、ダッシュボードからアクセス可能なストレージにメンテナンススケジュール情報を保存する方法などがあります。

バッチ ETL プロセスでの予測を使用してデータを変換して拡充する場合は、データバッチ全体に必要な時間を短縮するために、スループットを最大化することを目標にします。一方、オンライン予測のためにストリーミングデータを処理する場合は、各予測を（ほぼ）リアルタイムで受信するためにレイテンシを最小化することを目標にします。したがって、モデルの呼び出しがボトルネックになる可能性があります。

コアコンポーネント

このソリューションのバッチテストとストリーミングテストでは、次の 3 つの主要なテクノロジーが使用されます。

Dataflow で稼働し、データを処理する Apache Beam。
ML モデルを実装してトレーニングする TensorFlow。
トレーニングされた ML モデルでのバッチ予測とオンライン予測の実行にホスティングプラットフォームとして一部のテストで使用される AI Platform。

このソリューションでは、以下の理由によりデータパイプラインの実行に、Dataflow 上で動作する Apache Beam を選択しました。

Apache Beam は、ストリーミングとバッチの両方のデータ処理を行うオープンソースの統合プログラミングモデルです。
Dataflow は、サーバーなしで Apache Beam ジョブを実行できる Google Cloud プロダクトです。

TensorFlow は、機械学習フレームワークとして使用される Google のオープンソースの数学ライブラリです。TensorFlow を使用すると、単一のマシンまたは分散環境でモデルの構築、トレーニング、提供を行えます。モデルは、さまざまなデバイスに移植でき、使用可能な CPU、GPU、TPU などのリソースをトレーニングやサービスに利用できます。

AI Platform は、ハイパーパラメータの調整機能を使用してトレーニング、調整を行えるサーバーレスプラットフォームであり、DevOps で必要な最小限の管理で大量の TensorFlow モデルを処理します。AI Platform は、トレーニングされたモデルをオンライン予測用の REST API としてのデプロイと、バッチ予測ジョブの送信をサポートしています。AI Platform は、モデルをマイクロサービスとして提供できるいくつかのオプションの 1 つです。

このソリューションで詳しく説明しているアプローチでは、データ処理パイプラインに Dataflow を使用し、モデルを HTTP エンドポイントとしてホストするために AI Platform を使用しています。ただし、これらのアプローチは他のテクノロジーに置換できます。HTTP と直接的な TensorFlow モデルのパフォーマンス比較が大幅に変化することはありません。

バッチデータの処理とストリーミングデータの処理

このソリューションのテストには、バッチとストリーミングの両方のユースケースが含まれます。各テストでは、無制限のソースと制限付きのソースでは操作要件が異なるため、入力と出力に異なる Google Cloud プロダクトを利用しています。

制限付きデータセットのバッチ処理

図 1 は、典型的なバッチ処理パイプラインでは、生の入力データが Cloud Storage などのオブジェクトストレージに格納されることを示しています。構造化データストレージ形式には、カンマ区切り値（CSV）、最適化された行列（ORC）、Parquet、または Avro があります。これらの形式は、データがデータベースまたはログから取得される場合によく使用されます。

一般的なバッチ処理パイプラインのアーキテクチャ — **図 1.** バッチ処理アーキテクチャ

BigQuery などの分析プラットフォームの中には、クエリ機能に加えてストレージを提供するものもあります。BigQuery ではストレージ用に Capacitor を使用します。Dataflow 上の Apache Beam では、バッチ処理パイプラインの他のストレージオプションに加えて、BigQuery と Cloud Storage の両方から読み書きできます。

無制限のデータストリームのストリーム処理

ストリーミングの場合、データ処理パイプラインへの入力は通常、図 2 に示すようにメッセージングシステムです。Pub/Sub や Kafka などのテクノロジーは、通常、JSON、CSV、protobuf などの形式の個々のデータポイントを取り込むために使用されます。

一般的なストリーム処理パイプラインのアーキテクチャ — **図 2.** ストリーム処理アーキテクチャ

データポイントは、ウィンドウ処理機能を使用して一時的なイベント処理を実行することによって、個別に、またはマイクロバッチでグループとして処理できます。処理されたデータは、以下を含めたいくつかの宛先に送信できます。

ストリーミング API によるアドホック分析用の BigQuery。
リアルタイム情報を提供するための Bigtable。
後続のプロセスやパイプラインをトリガーするための Pub/Sub トピック。

制限付きと無制限の両方のデータソースシンクのソースコネクタ（入力）とシンクコネクタ（出力）の完全なリストは、Apache Beam I/O ページにあります。

TensorFlow モデルの呼び出し

TensorFlow トレーニングモデルは、次の 3 つの方法で呼び出せます。

オンライン予測用の HTTP エンドポイントを介して呼び出す。
バッチ予測とオンライン予測の際に、保存されたモデルファイルを使用して直接呼び出す。
バッチ予測の際に AI Platform バッチ予測ジョブで呼び出す。

オンライン予測用の HTTP エンドポイント

TensorFlow モデルは、ストリームデータ処理パイプラインまたはクライアントアプリを介して呼び出される HTTP エンドポイントとしてデプロイされ、リアルタイムで予測を行います。

TensorFlow モデルは、TensorFlow Serving や他の Seldon などのホスティングサービスを使用して、オンライン予測の HTTP エンドポイントとしてデプロイできます。図 3 に示すように、次のオプションのいずれかを選択できます。

1 つまたは複数の Compute Engine インスタンスにモデルをデプロイします。
Compute Engine または Google Kubernetes Engine で Docker イメージを使用します。
Kubeflow を利用して、Kubernetes または Google Kubernetes Engine へのデプロイを容易にします。
App Engine をエンドポイントとともに使用して、ウェブアプリでモデルをホストします。
フルマネージド ML トレーニングと Google Cloud 上でのサービス提供を行う AI Platform を使用します。

モデルを HTTP エンドポイントとして提供する Dataflow のオプション — **図 3.** モデルを HTTP エンドポイントとして提供する Dataflow の各種のオプション

AI Platform はフルマネージド型のサービスであるため、他のオプションより実装が簡単です。このため、このテストでは、モデルを HTTP エンドポイントとして提供するためのオプションとして、このサービスを使用します。これにより、さまざまな HTTP モデル提供オプションを比較するのではなく、直接モデルと AI Platform の HTTP エンドポイントとのパフォーマンスの比較を中心に扱うことができます。

AI Platform 予測によるオンライン予測の提供

オンライン予測を提供するには、次の 2 つのタスクを実行する必要があります。

モデルのデプロイ。
デプロイされたモデルとの推論のための対話（つまり、予測の実施）。

AI Platform を使用して HTTP エンドポイントとしてモデルをデプロイするには、次の手順が必要です。

トレーニング済みのモデルファイルを Cloud Storage で使用できることを確認します。
gcloud ml-engine models create コマンドを使用してモデルを作成します。
Cloud Storage のモデルファイルとともに、gcloud ml-engine versions create コマンドを使用してモデルバージョンをデプロイします。

モデルをデプロイするには、次のようなコマンドを使用します。

blogs/tf_dataflow_serving/scripts/deploy-cmle.sh

GitHub で表示


PROJECT="[PROJECT_ID]" # change to your project name
REGION="[REGION]"
BUCKET="[BUCKET]" # change to your bucket name
MODEL_NAME="babyweight_estimator" # change to your estimator name
MODEL_VERSION="v1" # change to your model version
MODEL_BINARIES=gs://${BUCKET}/models/${MODEL_NAME}

# upload the local SavedModel to GCS
gsutil -m cp -r model/trained/v1/* gs://${BUCKET}/models/${MODEL_NAME}

# set the current project
gcloud config set project ${PROJECT}

# list model files on GCS
gsutil ls ${MODEL_BINARIES}

# deploy model to GCP
gcloud ml-engine models create ${MODEL_NAME} --regions=${REGION}

# deploy model version
gcloud ml-engine versions create ${MODEL_VERSION} --model=${MODEL_NAME} --origin=${MODEL_BINARIES} --runtime-version=1.4

このコードは、モデルバージョン v1 で、Google Cloud プロジェクトに babyweight_estimator という AI Platform Prediction モデルを作成します。

モデルのデプロイ後は、そのモデルを呼び出せるようになります。次の Python コードは、AI Platform でモデルバージョンを REST API として呼び出す方法を示しています。

blogs/tf_dataflow_serving/model/inference.py

GitHub で表示

cmle_api = None

def init_api():

    global cmle_api

    if cmle_api is None:
        cmle_api = discovery.build('ml', 'v1',
                              discoveryServiceUrl='https://storage.googleapis.com/cloud-ml/discovery/ml_v1_discovery.json',
                              cache_discovery=True)

def estimate_cmle(instances):
    """
    Calls the babyweight estimator API on CMLE to get predictions

    Args:
       instances: list of json objects
    Returns:
        int - estimated baby weight
    """
    init_api()

    request_data = {'instances': instances}

    model_url = 'projects/{}/models/{}/versions/{}'.format(PROJECT, CMLE_MODEL_NAME, CMLE_MODEL_VERSION)
    response = cmle_api.projects().predict(body=request_data, name=model_url).execute()
    values = [item["predictions"][0] for item in response['predictions']]
    return values

BigQuery や Cloud Storage などの大規模なデータセットがあり、プロセス全体のスループットを最大化する場合、バッチ予測用に ML モデルを HTTP エンドポイントとして提供することは推奨されていません。これにより、データポイントごとに 1 つの HTTP リクエストが生成され、膨大な量の HTTP リクエストが発生します。次のセクションでは、バッチ予測により適したオプションを示します。

バッチ予測とオンライン予測のための直接モデル

直接モデル予測テクノロジーでは、Dataflow インスタンスのローカル TensorFlow SavedModel を利用します。保存されたモデルは、TensorFlow モデルの構築とトレーニングを完了した後に作成された出力ファイルのコピーです。TensorFlow SavedModel は、次のいずれかです。

Dataflow ジョブとして送信されるパイプラインソースコードの一部。
図 4 に示すように、Cloud Storage からダウンロードされた内容。

このソリューションでは、GitHub のソースコードの一部である SavedModel を使用しています。インスタンスにモデルを読み込むには、次のように行います。

Dataflow ジョブを作成するときに、モデルファイルを含めて、読み込むパイプラインの依存関係を指定します。次の Python コードは、Dataflow ジョブと一緒に送信されるモデルファイルが含まれる setup.py ファイルを示しています。

blogs/tf_dataflow_serving/setup.py

GitHub で表示
```
import setuptools

requirements = []

setuptools.setup(
    name='TF-DATAFLOW-DEMO',
    version='v1',
    install_requires=requirements,
    packages=setuptools.find_packages(),
    package_data={'model': ['trained/*',
                            'trained/v1/*',
                            'trained/v1/variables/*']
                  },
)
```

パイプラインでローカルモデルファイルを呼び出します。これにより、指定されたインスタンスの予測が生成されます。以下の Python コードにその方法を示しています。

blogs/tf_dataflow_serving/model/inference.py

GitHub で表示

predictor_fn = None

def init_predictor():
    """ Loads the TensorFlow saved model to the predictor object

    Returns:
        predictor_fn
    """

    global predictor_fn

    if predictor_fn is None:

        logging.info("Initialising predictor...")
        dir_path = os.path.dirname(os.path.realpath(__file__))
        export_dir = os.path.join(dir_path, SAVED_MODEL_DIR)

        if os.path.exists(export_dir):
            predictor_fn = tf.contrib.predictor.from_saved_model(
                export_dir=export_dir,
                signature_def_key="predict"
            )
        else:
            logging.error("Model not found! - Invalid model path: {}".format(export_dir))

def estimate_local(instances):
    """
    Calls the local babyweight estimator to get predictions

    Args:
       instances: list of json objects
    Returns:
        int - estimated baby weight
    """

    init_predictor()

    inputs = dict((k, [v]) for k, v in instances[0].items())
    for i in range(1,len(instances)):
        instance = instances[i]

        for k, v in instance.items():
            inputs[k] += [v]

    values = predictor_fn(inputs)['predictions']
    return [value.item() for value in values.reshape(-1)]

詳細については、Apache Beam の複数ファイルの依存関係ページをご覧ください。

AI Platform バッチ予測ジョブ

モデルを HTTP エンドポイントとしてデプロイする以外に、AI Platform では、デプロイされたモデルバージョンや Cloud Storage の TensorFlow SavedModel を使用してバッチ予測ジョブを実行できます。

AI Platform のバッチ予測ジョブでは、入力データファイルの Cloud Storage のロケーションがパラメータとして渡されます。モデルを使用してそのデータの予測を取得し、予測結果をパラメータとして指定された Cloud Storage の別の出力場所に保存します。次の例は、AI Platform バッチ予測ジョブを送信する gcloud コマンドを示しています。

blogs/tf_dataflow_serving/scripts/predict-batch-cmle.sh

GitHub で表示

BUCKET='<BUCKET>'
DATA_FORMAT="TEXT"
INPUT_PATHS=gs://${BUCKET}/data/babyweight/experiments/outputs/data-prep-*
OUTPUT_PATH=gs://${BUCKET}/data/babyweight/experiments/outputs/cmle-estimates
MODEL_NAME='babyweight_estimator'
VERSION_NAME='v1'
REGION='<REGION>'
now=$(date +"%Y%m%d_%H%M%S")
JOB_NAME="batch_predict_$MODEL_NAME$now"
MAX_WORKER_COUNT="20"

gcloud ml-engine jobs submit prediction $JOB_NAME \
    --model=$MODEL_NAME \
    --input-paths=$INPUT_PATHS \
    --output-path=$OUTPUT_PATH \
    --region=$REGION \
    --data-format=$DATA_FORMAT \
    --max-worker-count=$MAX_WORKER_COUNT

ポイントごとのオンライン予測とマイクロバッチのオンライン予測

リアルタイム予測パイプラインでは、モデルを HTTP エンドポイントとして提供している場合でも、モデルをワーカーから直接使用している場合でも、着信データポイントの予測を取得する 2 つのオプションがあります。

個々のポイント。この基本的なオプションでは、各データポイントを個別にモデルに送信して予測を得ます。
マイクロバッチ。この最適化オプションでは、ウィンドウ処理機能を使用して、特定の期間内（たとえば、5 秒ごと）にデータポイントをグループ化する、マイクロバッチを作成します。その後、マイクロバッチをモデルに送信して、すべてのインスタンスの予測を同時に取得します。

次の Python コードは、Apache Beam パイプラインでウィンドウ処理機能を使用して時間ベースのマイクロバッチを作成する方法を示しています。

blogs/tf_dataflow_serving/pipelines/stream_process.py

GitHub で表示

def run_pipeline_with_micro_batches(inference_type, project,
                                    pubsub_topic, pubsub_subscription,
                                    bq_dataset, bq_table,
                                    window_size, runner, args=None):

    prepare_steaming_source(project, pubsub_topic, pubsub_subscription)
    prepare_steaming_sink(project, bq_dataset, bq_table)
    pubsub_subscription_url = "projects/{}/subscriptions/{}".format(project, pubsub_subscription)
    options = beam.pipeline.PipelineOptions(flags=[], **args)

    pipeline = beam.Pipeline(runner, options=options)
    (
            pipeline
            | 'Read from PubSub' >> beam.io.ReadStringsFromPubSub(subscription=pubsub_subscription_url, id_label="source_id")
            | 'Micro-batch - Window Size: {} Seconds'.format(window_size) >> beam.WindowInto(FixedWindows(size=window_size))
            | 'Estimate Targets - {}'.format(inference_type) >> beam.FlatMap(lambda messages: estimate(messages, inference_type))
            | 'Write to BigQuery' >> beam.io.WriteToBigQuery(project=project,
                                                             dataset=bq_dataset,
                                                             table=bq_table
                                                             )
    )

    pipeline.run()

マイクロバッチ処理のアプローチでは、HTTP エンドポイントとしてデプロイされたモデルを使用します。これにより、HTTP リクエスト数が大幅に削減され、レイテンシが短縮されます。直接モデルで使用されるマイクロバッチ処理方法であっても、演算はベクトル化されているため、長さが 1 のテンソルを送信するよりも、N 個のインスタンスがあるテンソルを予測用に送信する方が効率的です。

バッチテスト

バッチテストでは、TensorFlow 回帰モデルを使用して BigQuery の Natality データセットで赤ちゃんの体重を予測します。Dataflow のバッチパイプラインを使用して、Cloud Storage の予測結果を CSV ファイルとして保存する必要があります。次のセクションでは、この作業を達成するために試したさまざまなテストについて説明します。

アプローチ 1: 直接モデル予測を使用した Dataflow

このアプローチでは、Dataflow ワーカーが、各レコードのバッチ処理パイプラインで予測のために直接呼び出される TensorFlow SavedModel をホストします。図 5 は、このアプローチのアーキテクチャの概要を示しています。

Dataflow パイプラインでは、次の手順に従います。

BigQuery からデータを読み取ります。
予測用の BigQuery レコードを準備します。
各レコードの予測を取得するために、ローカルの TensorFlow SavedModel を呼び出します。
結果（入力レコードと予測された赤ちゃんの体重）を CSV ファイルに変換します。
CSV ファイルを Cloud Storage に書き込みます。

このアプローチでは、HTTP エンドポイントとして AI Platform にデプロイされたモデルなど、リモートサービスへの呼び出しはありません。予測は、TensorFlow SavedModel を使用して各 Dataflow ワーカー内でローカルに行われます。

アプローチ 2: AI Platform バッチ予測を使用した Dataflow

このアプローチでは、TensorFlow SavedModel が Cloud Storage に格納され、AI Platform で予測に使用されます。ただし、前述のアプローチのように各レコードについて、デプロイされたモデルへの API 呼び出しを行う代わりに、予測用にデータが準備され、バッチとして送信されます。

このアプローチには 2 つのフェーズがあります。

Dataflow は BigQuery から予測用のデータを作成し、そのデータを Cloud Storage に格納します。
AI Platform バッチ予測ジョブは作成されたデータとともに送信され、予測結果は Cloud Storage に格納されます。

図 6 に、この 2 フェーズのアプローチのアーキテクチャ全体を示します。

**図 6.** バッチアプローチ 2: AI Platform バッチ予測を使用した Dataflow

Dataflow パイプラインを含めたワークフローの手順は次のとおりです。

BigQuery からデータを読み取ります。
予測用の BigQuery レコードを準備します。
Cloud Storage に JSON データを書き込みます。モデル内の serving_fn 関数では、JSON インスタンスを入力として想定しています。
Cloud Storage で作成したデータとともに AI Platform バッチ予測ジョブを送信します。このジョブでは、予測結果も Cloud Storage に書き込みます。

Dataflow ジョブでは、AI Platform の予測ジョブが送信されるのではなく、予測のためのデータが作成されます。つまり、データ準備タスクとバッチ予測タスクは密結合されていません。Cloud Functions、Airflow、または任意のスケジューラでは、Dataflow ジョブを実行してから、AI Platform ジョブを送信してバッチ予測を行うことによって、ワークフローをオーケストレートできます。

データが次のケースに該当する場合、パフォーマンスと使いやすさの両方のために、AI Platform のバッチ予測をおすすめします。

以前にデータの取り込みプロセスが実行されたため、Cloud Storage に予測用の形式でデータが存在する。
ワークフローの第 1 フェーズを制御できない。たとえば、Cloud Storage に保存される予測用データを準備する Dataflow パイプラインを制御できない場合です。

テスト構成

3 回のテストで以下の構成を使用しました。

データサイズ: 10K、100K、1M、および 10M 行
Cloud Storage クラス: Regional Storage
Cloud Storage のロケーション: europe-west1-b
Dataflow のリージョン: europe-west1-b
Dataflow ワーカーのマシンタイプ: n1-standard-1
最大 100 万レコードまでのバッチデータ用の Dataflow の自動スケーリング
Dataflow num_worker: バッチデータの場合は 20、最大 1,000 万レコード
AI Platform バッチ予測 max-worker-count の設定: 20

Cloud Storage のロケーションと Dataflow リージョンは同じにする必要があります。このソリューションでは、任意の値として europe-west1-b リージョンを使用します。

結果

次の表は、さまざまなサイズのデータセットでバッチ予測と直接モデル予測を行った結果（タイミング）をまとめたものです。

バッチデータサイズ	指標	Dataflow、AI Platform バッチ予測	直接モデル予測を使用した Dataflow
1 万行	実行時間	15 分 30 秒（Dataflow: 7 分 47 秒 + AI Platform: 7 分 43 秒）	8 分 24 秒
	合計 vCPU 時間	0.301 時間（Dataflow: 0.151 時間 + AI Platform: 0.15 時間）	0.173 時間
10 万行	実行時間	16 分 37 秒（Dataflow: 8 分 39 秒 + AI Platform: 7 分 58 秒）	10 分 31 秒
	合計 vCPU 時間	0.334 時間（Dataflow: 0.184 時間 + AI Platform: 0.15 時間）	0.243 時間
100 万行	実行時間	21 分 11 秒（Dataflow: 11 分 7 秒 + AI Platform: 10 分 4 秒）	17 分 12 秒
	合計 vCPU 時間	0.446 時間（Dataflow: 0.256 時間 + AI Platform: 0.19 時間）	1.115 時間
1,000 万行	実行時間	33 分 08 秒（Dataflow: 12 分 15 秒 + AI Platform: 20 分 53 秒）	25 分 02 秒
	合計 vCPU 時間	5.251 時間（Dataflow: 3.581 時間 + AI Platform: 1.67 時間）	7.878 時間

図 7 にこれらの結果のグラフを示します。

**図 7.** 4 つの異なるデータセットサイズに対する 3 つのアプローチのタイミングを示すグラフ

この結果が示すように、AI Platform のバッチ予測ジョブは、データがすでに予測に使用される形式で Cloud Storage に存在する場合、入力データに対する予測の生成に要する時間が短くなります。ただし、バッチ予測ジョブを前処理ステップ（予測データの BigQuery から Cloud Storage への抽出と作成）および後処理ステップ（データを BigQuery に保存する）と組み合わせると、直接モデルのアプローチのほうが全体的な実行時間は短くなります。さらに、（ストリーミングテストに関して後で説明する）マイクロバッチ処理を使用すると、直接モデル予測アプローチのパフォーマンスをさらに改善することができます。

ストリームテスト

ストリーミングテストでは、Dataflow パイプラインで Pub/Sub トピックからデータポイントを読み取り、ストリーミング API を使用して BigQuery にデータを書き込みます。Dataflow ストリーミングパイプラインでは、TensorFlow の赤ちゃんの体重の推定モデルを使用してデータを処理し、予測を取得します。

このトピックでは、データポイントを生成するストリームシミュレータからデータを受け取ります。データポイントは、事前に定義された毎秒のイベントレートで、赤ちゃんの体重を予測するインスタンスです。これは、無制限のデータソースの実際の例をシミュレートします。次の Python コードは、Pub/Sub トピックに送信されるデータストリームをシミュレートします。

blogs/tf_dataflow_serving/simulate_stream.py

GitHub で表示

client = pubsub.Client(project=PARAMS.project_id)
topic = client.topic(PARAMS.pubsub_topic)
if not topic.exists():
    print 'Topic does not exist. Please run a stream pipeline first to create the topic.'
    print 'Simulation aborted.'

    return

for index in range(PARAMS.stream_sample_size):

    message = send_message(topic, index)

    # for debugging
    if PARAMS.show_message:
        print "Message {} was sent: {}".format(index+1, message)
        print ""

    time.sleep(sleep_time_per_msg)

アプローチ 1: AI Platform オンライン予測を使用した Dataflow

このアプローチでは、TensorFlow モデルが、AI Platform で REST API としてデプロイされてホストされます。Dataflow ストリーミングパイプラインでは、Pub/Sub で取得した予測の各メッセージの API を呼び出します。図 8 は、このアプローチのアーキテクチャの概要を示しています。

このアプローチでは、Dataflow パイプラインで次の手順に従います。

Pub/Sub トピックからメッセージを読み込みます。
AI Platform モデルの API に HTTP リクエストを送信して、各メッセージの予測を取得します。
ストリーミング API を使用して BigQuery に結果を書き込みます。

マイクロバッチ処理はより優れたアプローチです。つまり、Pub/Sub から読み取られる各メッセージについてモデルの REST API に対して HTTP リクエストを送信する代わりに、Dataflow では 1 秒間に受信したメッセージをグループ化します。次に、このグループのメッセージを、単一の HTTP リクエスト内のマイクロバッチとしてモデルの API に送信します。このアプローチでは、Dataflow パイプラインで次の手順に従います。

Pub/Sub トピックからメッセージを読み込みます。
1 秒間のウィンドウ処理オペレーションを適用して、メッセージのマイクロバッチを作成します。
HTTP リクエストをマイクロバッチとともに AI Platform モデルの API に送信して、メッセージの予測を取得します。
ストリーミング API を使用して BigQuery に結果を書き込みます。

このアプローチの理論的基礎は以下のとおりです。

AI Platform モデルなどのリモートサービスへの呼び出し数を減らします。
各メッセージを処理する平均レイテンシを短縮します。
パイプライン全体の処理時間を短縮します。

このテストでは、時間枠を 1 秒に設定しました。ただし、マイクロバッチサイズ（AI Platform モードにバッチとして送信されるメッセージ数）はさまざまです。マイクロバッチサイズは、メッセージの生成頻度（1 秒あたりのメッセージ数）によって異なります。

次のセクションでは、メッセージ数が毎秒 50、100、500 の 3 つの異なる頻度でのテストについて説明します。つまり、マイクロバッチサイズは 50、100、500 です。

アプローチ 2: 直接モデル予測を使用した Dataflow

このアプローチは、バッチテストで使用されたアプローチと同様です。TensorFlow SavedModel は Dataflow ワーカーでホストされ、各レコードのストリーム処理パイプラインでの予測のために呼び出されます。図 9 は、このアプローチのアーキテクチャの概要を示しています。

このアプローチでは、Dataflow パイプラインで次の手順に従います。

Pub/Sub トピックからメッセージを読み込みます。
各レコードの予測を取得するために、ローカルの TensorFlow SavedModel を呼び出します。
ストリーミング API を使用して BigQuery に結果を書き込みます。

マイクロバッチ処理テクノロジーは、直接モデル予測アプローチを使用したストリームパイプラインでも使用できます。1 つのデータインスタンスのテンソルをモデルに送信する代わりに、N 個のデータインスタンスのテンソルを送信できます。ここで、N は Dataflow の時間枠内にモデルで受信したメッセージ数です。このアプローチでは、TensorFlow モデルのベクトル化された演算を使用して、いくつかの予測を並行して取得します。

テスト構成

これらのテストには次の構成を使用しました。

ストリームデータサイズ: 10K records (messages)
シミュレートされた 1 秒あたりのメッセージ数（MPS）: 50、100、500
時間枠のサイズ（マイクロバッチテスト）: 1 second
Dataflow のリージョン: europe-west1-b
Dataflow ワーカーのマシンタイプ: n1-standard-1
Dataflow num_worker: 5（自動スケーリングなし）
AI Platform モデルの API ノード: 3 (manualScale)

結果

次の表は、さまざまな量のデータ（1 秒あたりのメッセージ数）でストリーミングテストを実行した結果をまとめたものです。メッセージの頻度とは、1 秒あたりに送信されるメッセージ数を意味し、シミュレーション時間とは、すべてのメッセージを送信するための時間を意味します。

メッセージのストリーミング頻度	指標	AI Platform オンライン予測を使用した Dataflow		直接モデル予測を使用した Dataflow
		単一のメッセージ	マイクロバッチ処理	単一のメッセージ	マイクロバッチ処理
毎秒 50 メッセージ（シミュレーション時間: 3 分 20 秒）	合計時間	9 分 34 秒	7 分 44 秒	3 分 43 秒	3 分 22 秒
毎秒 100 メッセージ（シミュレーション時間: 1 分 40 秒）	合計時間	6 分 03 秒	4 分 34 秒	1 分 51 秒	1 分 41 秒
毎秒 500 メッセージ（シミュレーション時間: 20 秒）	合計時間	NA - デフォルトの AI Platform オンライン予測割り当て	2 分 47 秒	1 分 23 秒	48 秒

図 10 にこれらの結果のグラフを示します。

結果に示すように、マイクロバッチ処理テクノロジーは、AI Platform のオンライン予測と直接モデル予測の双方で実行パフォーマンスを向上させます。さらに、ストリーミングパイプラインで直接モデルを使用すると、オンライン予測で外部 REST/HTTP API を呼び出す場合と比べて、パフォーマンスが 2 倍から 4 倍向上します。

まとめ

説明したアプローチとテスト結果に従って、以下のアプローチをおすすめします。

バッチ処理

バッチデータ処理パイプラインを構築し、パイプラインの一部として予測を行う場合、最高のパフォーマンスを得るには直接モデルアプローチを使用します。
予測のローカルモデルを呼び出す前にデータポイントのマイクロバッチを作成して、ベクトル化された操作の並列化を利用することによって、直接モデルアプローチのパフォーマンスを向上させます。
予測に適した形式でデータが Cloud Storage に入力されている場合、AI Platform のバッチ予測を使用することで最高のパフォーマンスを得ることができます。
バッチ予測に GPU の能力を使用したい場合は、AI Platform を使用します。
バッチ予測には AI Platform オンライン予測を使用しないでください。

ストリーム処理

最高のパフォーマンスを実現し、平均レイテンシを短縮するには、ストリーミングパイプラインで直接モデルを使用します。予測はローカルで実施され、リモートサービスへの HTTP 呼び出しは行われません。
モデルをデータ処理パイプラインから切り離すことで、オンライン予測で使用されるモデルの保守性が向上します。最良のアプローチは、AI Platform またはその他のウェブホスティングサービスを使用して、モデルを独立したマイクロサービスとして提供することです。
モデルを独立したウェブサービスとしてデプロイすることで、複数のデータ処理パイプラインとオンラインアプリはモデルサービスをエンドポイントとして使用できます。モデルに対する変更は、そのモデルを使用するアプリやパイプラインに対して透過的です。
負荷分散を使用して複数のサービスインスタンスをデプロイし、モデルのウェブサービスのスケーラビリティと可用性を改善します。AI Platform では、モデルバージョンをデプロイするときに必要なのは、yaml 構成ファイル内のノード数（manualScaling）または minNodes（autoScaling）を指定することだけです。
モデルを別のマイクロサービスにデプロイする場合、基盤となるサービスインフラストラクチャに応じて追加コストが発生します。AI Platform のオンライン予測については、料金に関するよくある質問をご覧ください。
直接モデルと HTTP モデルの両方のサービスでパフォーマンスを向上させるには、ストリーミングデータ処理パイプラインでマイクロバッチ処理を使用します。マイクロバッチ処理は、モデルサービスへの HTTP リクエストの数を減らし、TensorFlow モデルのベクトル化された演算を使用して予測を取得します。

次のステップ

構造化データによる機械学習ソリューションで赤ちゃんの体重モデルを構築し、トレーニングする方法を習得する。
GitHub のコンパニオンリポジトリを参照する。
Google Cloud に関するリファレンスアーキテクチャ、図、チュートリアル、ベストプラクティスを確認する。Cloud Architecture Center を確認します。

Cloud Dataflow パイプラインでの予測のための機械学習モデルの比較

概要

プラットフォーム

コア コンポーネント

バッチデータの処理とストリーミング データの処理

制限付きデータセットのバッチ処理

無制限のデータ ストリームのストリーム処理

TensorFlow モデルの呼び出し

オンライン予測用の HTTP エンドポイント

AI Platform 予測によるオンライン予測の提供

バッチ予測とオンライン予測のための直接モデル

AI Platform バッチ予測ジョブ

ポイントごとのオンライン予測とマイクロバッチのオンライン予測

バッチテスト

アプローチ 1: 直接モデル予測を使用した Dataflow

アプローチ 2: AI Platform バッチ予測を使用した Dataflow

テスト構成

結果

ストリーム テスト

アプローチ 1: AI Platform オンライン予測を使用した Dataflow

アプローチ 2: 直接モデル予測を使用した Dataflow

テスト構成

結果

まとめ

バッチ処理

ストリーム処理

次のステップ

コアコンポーネント

バッチデータの処理とストリーミングデータの処理

無制限のデータストリームのストリーム処理

ストリームテスト