Google Cloud での傾向モデリングに Kubeflow Pipelines を使用する

このシナリオの Jupyter ノートブック

パイプラインを作成および構築するタスクは、GitHub リポジトリにある Jupyter ノートブックに組み込まれています。

タスクを実施するには、ノートブックを入手し、ノートブック内のコードセルを順番に実行します。このドキュメントで説明するフローでは、Vertex AI Workbench でノートブックを実行していることを前提としています。

Vertex AI Workbench 環境を開く

まず、GitHub リポジトリのクローンを Vertex AI Workbench 環境に作成します。

1. Google Cloud コンソールで、ノートブックを作成するプロジェクトを選択します。

2. [Vertex AI Workbench] ページに移動します。

   [Vertex AI Workbench] ページに移動

3. [ユーザー管理のノートブック] タブで、[新しいノートブック] をクリックします。

4. ノートブック タイプのリストで、Python 3 ノートブックを選択します。

5. [新しいノートブック] ダイアログで [詳細オプション] をクリックし、[マシンタイプ] で、使用するマシンタイプを選択します。不明な場合は、[n1-standard-1 (1 cVPU, 3.75 GB RAM)] を選択します。

6. [作成] をクリックします。

   ノートブック環境が作成されるまでに少し時間がかかります。

7. ノートブックが作成されたら、ノートブックを選択し、[JupyterLab を開く] をクリックします。

   ブラウザで JupyterLab 環境が開きます。

8. ターミナルタブを起動するには、[File] > [New] > [Launcher] の順に選択します。

9. [Launcher] タブの [Terminal] アイコンをクリックします。

10. ターミナルで、mlops-on-gcp GitHub リポジトリのクローンを作成します。

    git clone https://github.com/GoogleCloudPlatform/cloud-for-marketing/
    

    コマンドが終了すると、ファイル ブラウザに cloud-for-marketing フォルダが表示されます。

ノートブックを構成する

ノートブックは、実行する前に、それを構成する必要があります。ノートブックには、パイプライン アーティファクトを格納する Cloud Storage バケットが必要なため、まずそのバケットを作成します。

1. ノートブックがパイプライン アーティファクトを保存できる Cloud Storage バケットを作成します。バケットの名前は、グローバルに一意にする必要があります。
2. cloud-for-marketing/marketing-analytics/predicting/kfp_pipeline/ フォルダで、Propensity_Pipeline.ipynb ノートブックを開きます。
3. ノートブックで、PROJECT_ID 変数の値を、パイプラインを実行する Google Cloud プロジェクトの ID に設定します。
4. BUCKET_NAME 変数の値を、作成したバケットの名前に設定します。

このドキュメントの残りの部分では、パイプラインの仕組みを理解するために重要なコード スニペットについて説明します。完全な実装については、GitHub リポジトリをご覧ください。

BigQuery ビューを作成する

パイプラインの最初のステップでは、各モデルの構築に使用される入力データを生成します。この Kubeflow Pipelines コンポーネントは、BigQuery ビューを生成します。ビューの作成プロセスを簡素化するため、一部の SQL はすでに生成され、テキスト ファイルで GitHub に保存されています。

各コンポーネントのコードは、Kubeflow Pipelines コンポーネント クラスの修飾（親クラスまたは属性による関数の変更）から始まります。このコードは、create_input_view 関数（パイプラインの 1 ステップ）を定義します。

関数にはいくつかの入力が必要です。現在、これらの値の一部は、コード内にハードコードされています（開始日と終了日など）。パイプラインを自動化する際には、適切な値を使用するようにコードを変更するか（たとえば、日付に CURRENT_DATE 関数を使用します）、これらの値をハードコードするのではなく、パラメータとして受け取るようにコンポーネントを更新することが可能です。また、ga_data_ref の値を GA360 テーブルの名前に変更し、conversion 変数の値を変換に設定する必要があります（この例では、一般公開されているアナリティクス 360 のサンプルデータを使用します）。

create-input-view コンポーネントのコードを以下に示します。

@component(
   # this component builds a BigQuery view, which will be the underlying source for model
   packages_to_install=["google-cloud-bigquery", "google-cloud-storage"],
   base_image="python:3.9",
   output_component_file="output_component/create_input_view.yaml",
)
def create_input_view(view_name: str,
                     data_set_id: str,
                     project_id: str,
                     bucket_name: str,
                     blob_path: str

):
   from google.cloud import bigquery
   from google.cloud import storage
   client = bigquery.Client(project=project_id)
   dataset = client.dataset(data_set_id)
   table_ref = dataset.table(view_name)
   ga_data_ref = 'bigquery-public-data.google_analytics_sample.ga_sessions_*'
   conversion = "hits.page.pageTitle like '%Shopping Cart%'"
   start_date = '20170101'
   end_date = '20170131'

def get_sql(bucket_name, blob_path):
       from google.cloud import storage
       storage_client = storage.Client()
       bucket = storage_client.get_bucket(bucket_name)
       blob = bucket.get_blob(blob_path)
       content = blob.download_as_string()
       return content
def if_tbl_exists(client, table_ref):

...

   else:
       content = get_sql()
       content = str(content, 'utf-8')
       create_base_feature_set_query = content.
                                   format(start_date = start_date,
                                   end_date = end_date,
                                   ga_data_ref = ga_data_ref,
                                   conversion = conversion)
shared_dataset_ref = client.dataset(data_set_id)
base_feature_set_view_ref = shared_dataset_ref.table(view_name)
base_feature_set_view = bigquery.Table(base_feature_set_view_ref)
base_feature_set_view.view_query = create_base_feature_set_query.format(project_id)
base_feature_set_view = client.create_table(base_feature_set_view)

BigQuery ML モデルを構築する

ビューを作成したら、build_bqml_logistic という名前のコンポーネントを実行して、BigQuery ML モデルを構築します。ノートブックのこのブロックは、コア コンポーネントです。最初のブロックで作成したトレーニング ビューを使用して、BigQuery ML モデルを構築します。この例では、ノートブックでロジスティック回帰を使用しています。

モデルタイプと使用可能なハイパーパラメータについては、BigQuery ML のリファレンス ドキュメントをご覧ください。

このコンポーネントのコードを以下に示します。

@component(
   # this component builds a logistic regression with BigQuery ML
   packages_to_install=["google-cloud-bigquery"],
   base_image="python:3.9",
   output_component_file="output_component/create_bqml_model_logistic.yaml"
)
def build_bqml_logistic(project_id: str,
                       data_set_id: str,
                       model_name: str,
                       training_view: str
):
   from google.cloud import bigquery
   client = bigquery.Client(project=project_id)
   model_name = f"{project_id}.{data_set_id}.{model_name}"
   training_set = f"{project_id}.{data_set_id}.{training_view}"
   build_model_query_bqml_logistic = '''
   CREATE OR REPLACE MODEL `{model_name}`
   OPTIONS(model_type='logistic_reg'
   , INPUT_LABEL_COLS = ['label']
   , L1_REG = 1
   , DATA_SPLIT_METHOD = 'RANDOM'
   , DATA_SPLIT_EVAL_FRACTION = 0.20
   ) AS
       SELECT * EXCEPT (fullVisitorId, label),
       CASE WHEN label is null then 0 ELSE label end as label
   FROM `{training_set}`
   '''.format(model_name = model_name, training_set = training_set)
job_config = bigquery.QueryJobConfig()
client.query(build_model_query_bqml_logistic, job_config=job_config)

BigQuery ML ではなく XGBoost を使用する

前のセクションで説明したコンポーネントは、BigQuery ML を使用しています。ノートブックの次のセクションでは、BigQuery ML を使用するのではなく Python で直接 XGBoost を使用する方法について説明します。

build_bqml_xgboost という名前のコンポーネントを実行して、グリッド検索を使用して標準の XGBoost 分類モデルを実行するコンポーネントを構築します。その後、作成した Cloud Storage バケットに、モデルをアーティファクトとして保存します。この関数では、出力アーティファクトの追加パラメータ（metrics と model）がサポートされており、これらは Kubeflow Pipelines で必須のパラメータです。

@component(
   # this component builds an xgboost classifier with xgboost
   packages_to_install=["google-cloud-bigquery", "xgboost", "pandas", "sklearn", "joblib", "pyarrow"],
   base_image="python:3.9",
   output_component_file="output_component/create_xgb_model_xgboost.yaml"
)
def build_xgb_xgboost(project_id: str,
                     data_set_id: str,
                     training_view: str,
                     metrics: Output[Metrics],
                     model: Output[Model]
):

...

  data_set = f"{project_id}.{data_set_id}.{training_view}"
  build_df_for_xgboost = '''
                         SELECT * FROM `{data_set}`
                         '''.format(data_set = data_set)

...

  xgb_model = XGBClassifier(n_estimators=50,
                            objective='binary:hinge',
                            silent=True,
                            nthread=1,
                           eval_metric="auc")
   random_search = RandomizedSearchCV(xgb_model,
                                     param_distributions=params,
                                     n_iter=param_comb,
                                     scoring='precision',
                                     n_jobs=4,
                                     cv=skf.split(X_train,y_train),
                                     verbose=3,
                                     random_state=1001 )
  random_search.fit(X_train, y_train)
  xgb_model_best = random_search.best_estimator_
  predictions = xgb_model_best.predict(X_test)
  score = accuracy_score(y_test, predictions)
  auc = roc_auc_score(y_test, predictions)
  precision_recall = precision_recall_curve(y_test, predictions)

  metrics.log_metric("accuracy",(score * 100.0))
  metrics.log_metric("framework", "xgboost")
  metrics.log_metric("dataset_size", len(df))
  metrics.log_metric("AUC", auc)

  dump(xgb_model_best, model.path + ".joblib")

エンドポイントを構築する

前のセクションの XGBoost モデルを使用して、deploy_xgb という名前のコンポーネントを実行し、エンドポイントを構築します。このコンポーネントは、前の XGBoost モデル アーティファクトを取得し、コンテナを構築してエンドポイントをデプロイします。同時に、エンドポイント URL をアーティファクトとして指定し、表示できるようにします。この手順が完了すると、Vertex AI エンドポイントが作成され、Vertex AI のコンソール ページにエンドポイントが表示されます。

@component(
   # Deploys xgboost model
   packages_to_install=["google-cloud-aiplatform", "joblib", "sklearn", "xgboost"],
   base_image="python:3.9",
   output_component_file="output_component/xgboost_deploy_component.yaml",
)
def deploy_xgb(
   model: Input[Model],
   project_id: str,
   vertex_endpoint: Output[Artifact],
   vertex_model: Output[Model]
):
   from google.cloud import aiplatform
   aiplatform.init(project=project_id)
   deployed_model = aiplatform.Model.upload(
       display_name="tai-propensity-test-pipeline",
       artifact_uri = model.uri.replace("model", ""),
       serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/xgboost-cpu.1-4:latest"
   )
   endpoint = deployed_model.deploy(machine_type="n1-standard-4")
# Save data to the output params
   vertex_endpoint.uri = endpoint.resource_name
   vertex_model.uri = deployed_model.resource_name

パイプラインを定義する

パイプラインを定義するには、前に作成したコンポーネントに基づいて各オペレーションを定義します。次に、コンポーネント内で明示的に呼び出されない場合は、パイプライン要素の順序を指定できます。

たとえば、ノートブックで次のコードはパイプラインを定義します。この場合、build_bqml_logistic_op コンポーネントを create_input_view_op コンポーネントの後に実行する必要があります。

@dsl.pipeline(
   # Default pipeline root. You can override it when submitting the pipeline.
   pipeline_root=PIPELINE_ROOT,
   # A name for the pipeline.
   name="pipeline-test",
   description='Propensity BigQuery ML Test'
)
def pipeline():

   create_input_view_op = create_input_view(
                          view_name = VIEW_NAME,
                          data_set_id = DATA_SET_ID,
                          project_id = PROJECT_ID,
                          bucket_name = BUCKET_NAME,
                          blob_path = BLOB_PATH
                                            )
    build_bqml_logistic_op = build_bqml_logistic(
                        project_id = PROJECT_ID,
                        data_set_id = DATA_SET_ID,
                        model_name = 'bqml_logistic_model',
                        training_view = VIEW_NAME
                                                  )

 # several components have been deleted for brevity

   build_bqml_logistic_op.after(create_input_view_op)
   build_bqml_xgboost_op.after(create_input_view_op)
   build_bqml_automl_op.after(create_input_view_op)
   build_xgb_xgboost_op.after(create_input_view_op)

   evaluate_bqml_logistic_op.after(build_bqml_logistic_op)
   evaluate_bqml_xgboost_op.after(build_bqml_xgboost_op)
   evaluate_bqml_automl_op.after(build_bqml_automl_op)

パイプラインをコンパイルして実行する

これでパイプラインをコンパイルして実行できるようになりました。

キャッシュを有効にするため、ノートブックの次のコードでは enable_caching 値を true に設定します。キャッシュ保存が有効になっている場合、コンポーネントが正常に完了する前の実行は再実行されません。このフラグは、特にパイプラインのテストの際に役立ちます。キャッシュが有効になっていると、実行が高速になり、使用するリソースの量は少なくなります。

compiler.Compiler().compile(
   pipeline_func=pipeline, package_path="pipeline.json"
)
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")
run = pipeline_jobs.PipelineJob(
   display_name="test-pipeine",
   template_path="pipeline.json",

   job_id="test-{0}".format(TIMESTAMP),
   enable_caching=True
)
run.run()

パイプラインを自動化する

この段階で、最初のパイプラインが起動されました。このジョブのステータスは、コンソールの [Vertex AI Pipelines] ページで確認できます。各コンテナがビルドおよび実行されることを確認できます。このセクションでは、特定のコンポーネントのエラーをクリックして、そのエラーを追跡することもできます。

パイプラインをスケジュールするには、Cloud Functions の関数を構築し、cron ジョブに似たスケジューラを使用します。

ノートブックの最後のセクションのコードは、次のコード スニペットに示すように、パイプラインが 1 日に 1 回実行されるようにスケジュールします。

from kfp.v2.google.client import AIPlatformClient
api_client = AIPlatformClient(project_id=PROJECT_ID,
                            region='us-central1'
                            )
api_client.create_schedule_from_job_spec(
   job_spec_path='pipeline.json',
   schedule='0 * * * *',
   enable_caching=False
)

完成したパイプラインを本番環境で使用する

完成したパイプラインは次のタスクを実行しました。

• 入力データセットを作成する
• BigQuery ML と Python の XGBoost の両方を使用して複数のモデルをトレーニングする。
• モデルの結果を分析する。
• XGBoost モデルをデプロイする。

また、Cloud Functions と Cloud Scheduler を使用して毎日実行し、パイプラインを自動化しました。

ノートブックに定義されているパイプラインは、さまざまなモデルの作成方法を示すために作られています。現在パイプラインは本番環境シナリオで構築されているため、パイプラインは実行しません。しかし、このパイプラインをガイドとして使用し、ニーズに合わせてコンポーネントを変更できます。たとえば、特徴作成プロセスを編集して、データの活用、期間の変更、別モデルの構築を行えます。また、本番環境の要件に最適なモデルも選択します。

パイプラインを本番環境で使用できるようになると、追加のタスクを実装できます。たとえば、チャンピオン / チャレンジャー モデルを実装できます。この場合、新しいモデルが作成され、新しいモデル（チャレンジャー）と既存のモデル（チャンピオン）の両方が毎日スコア付けされます。新しいモデルのパフォーマンスが、現在のモデルのパフォーマンスよりも優れている場合にのみ、本番環境に配置します。システムの進行状況をモニタリングするために、毎日のモデル パフォーマンスの記録を保存し、パフォーマンスの傾向を可視化することもできます。