Menggunakan Kubeflow Pipeline untuk pemodelan kecenderungan di Google Cloud

Notebook Jupyter untuk skenario ini

Tugas untuk membuat dan membangun pipeline di-build ke notebook Jupyter yang ada di repositori GitHub.

Untuk menjalankan tugas ini, dapatkan notebook lalu jalankan sel kode di notebook secara berurutan. Alur yang dijelaskan dalam dokumen ini mengasumsikan bahwa Anda menjalankan notebook di Vertex AI Workbench.

Membuka lingkungan Vertex AI Workbench

Anda dapat memulai dengan meng-clone repositori GitHub ke lingkungan Vertex AI Workbench.

1. Di konsol Google Cloud, pilih project tempat Anda ingin membuat notebook.

2. Buka halaman Vertex AI Workbench.

   Buka halaman Vertex AI Workbench

3. Pada tab Notebook yang dikelola pengguna, klik Notebook baru.

4. Dalam daftar jenis notebook, pilih notebook Python 3.

5. Pada dialog Notebook baru, klik Opsi lanjutan, lalu di bagian Jenis mesin, pilih jenis mesin yang ingin Anda gunakan. Jika Anda tidak yakin, pilih n1-standard-1 (1 cVPU, 3,75 GB RAM).

6. Klik Create.

   Proses pembuatan lingkungan notebook memerlukan waktu beberapa saat.

7. Setelah notebook dibuat, pilih notebook, lalu klik Buka Jupyterlab.

   Lingkungan JupyterLab akan terbuka di browser Anda.

8. Untuk meluncurkan tab terminal, pilih File > Baru > Peluncur.

9. Klik ikon Terminal pada tab Peluncur.

10. Di terminal, lakukan clone repositori GitHub mlops-on-gcp:

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

    Setelah perintah selesai, Anda akan melihat folder cloud-for-marketing di file browser.

Mengonfigurasi setelan notebook

Sebelum menjalankan notebook, Anda harus mengonfigurasinya. Notebook ini memerlukan bucket Cloud Storage untuk menyimpan artefak pipeline, jadi, untuk memulai Anda harus membuat bucket tersebut.

1. Buat bucket Cloud Storage tempat notebook dapat menyimpan artefak pipeline. Nama bucket harus unik secara global.
2. Pada folder cloud-for-marketing/marketing-analytics/predicting/kfp_pipeline/, buka notebook Propensity_Pipeline.ipynb.
3. Di notebook, tetapkan nilai variabel PROJECT_ID ke ID project Google Cloud tempat Anda ingin menjalankan pipeline.
4. Tetapkan nilai variabel BUCKET_NAME ke nama bucket yang baru saja Anda buat.

Bagian selanjutnya dari dokumen ini menjelaskan cuplikan kode yang penting untuk memahami cara kerja pipeline. Untuk mengetahui implementasi yang lengkap, lihat repositori GitHub.

Membangun tampilan BigQuery

Langkah pertama dalam pipeline menghasilkan data input, yang akan digunakan untuk membangun setiap model. Komponen Kubeflow Pipelines ini menghasilkan tampilan BigQuery. Untuk menyederhanakan proses pembuatan tampilan, beberapa SQL telah dibuat dan disimpan dalam file teks di GitHub.

Kode untuk setiap komponen dimulai dengan mendekorasi (mengubah kelas atau fungsi induk melalui atribut) kelas komponen Kubeflow Pipelines. Kode tersebut kemudian menentukan fungsi create_input_view, yang merupakan langkah dalam pipeline.

Fungsi ini memerlukan beberapa input. Beberapa nilai tersebut saat ini di-hard code ke dalam kode, seperti tanggal mulai dan tanggal akhir. Saat mengotomatiskan pipeline, Anda dapat mengubah kode untuk menggunakan nilai yang sesuai (misalnya, menggunakan fungsi CURRENT_DATE untuk suatu tanggal), atau Anda dapat mengupdate komponen untuk menggunakan nilai ini sebagai parameter, bukan menyimpannya secara hard code. Anda juga harus mengubah nilai ga_data_ref menjadi nama tabel GA360, dan menetapkan nilai variabel conversion ke konversi Anda. (Contoh ini menggunakan data sampel GA360 publik.)

Listingan berikut menunjukkan kode untuk komponen 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)

Membangun model BigQuery ML

Setelah tampilan dibuat, jalankan komponen bernama build_bqml_logistic untuk membangun model BigQuery ML. Blok notebook ini adalah komponen inti. Dengan menggunakan tampilan pelatihan yang Anda buat di blok pertama, model ini akan membangun model BigQuery ML. Dalam contoh ini, notebook menggunakan regresi logistik.

Untuk mendapatkan informasi tentang jenis model dan hyperparameter yang tersedia, baca Dokumentasi referensi ML BigQuery.

Listingan berikut menunjukkan kode untuk komponen ini.

@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)

Menggunakan XGBoost, bukan BigQuery ML

Komponen yang diilustrasikan di bagian sebelumnya menggunakan BigQuery ML. Bagian notebook berikut ini menunjukkan cara menggunakan XGBoost di Python secara langsung, bukan menggunakan BigQuery ML.

Anda menjalankan komponen bernama build_bqml_xgboost untuk membuat komponen sehingga dapat menjalankan model klasifikasi XGBoost standar dengan penelusuran petak. Kode tersebut kemudian menyimpan model sebagai artefak di bucket Cloud Storage yang Anda buat. Fungsi ini mendukung parameter tambahan (metrics dan model) untuk artefak output; parameter-parameter ini diperlukan oleh 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")

Membangun endpoint

Anda menjalankan komponen bernama deploy_xgb untuk membangun endpoint menggunakan model XGBoost dari bagian sebelumnya. Komponen ini mengambil artefak model XGBoost sebelumnya, membangun container, lalu men-deploy endpoint, sekaligus menyediakan URL endpoint sebagai artefak agar Anda dapat melihatnya. Setelah langkah ini selesai, endpoint Vertex AI telah dibuat dan Anda dapat melihat endpoint tersebut di halaman konsol untuk 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

Menentukan pipeline

Untuk menentukan pipeline, tentukan setiap operasi berdasarkan komponen yang Anda buat sebelumnya. Kemudian, Anda dapat menentukan urutan elemen pipeline jika tidak dipanggil secara eksplisit dalam komponen.

Misalnya, kode di notebook berikut ini mendefinisikan pipeline. Dalam hal ini, kode memerlukan komponen build_bqml_logistic_op untuk dijalankan setelah komponen 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)

Mengompilasi dan menjalankan pipeline

Sekarang Anda dapat mengompilasi dan menjalankan pipeline.

Kode berikut di notebook menetapkan nilai enable_caching ke benar (true) untuk mengaktifkan cache. Ketika caching diaktifkan, setiap operasi sebelumnya saat komponen berhasil diselesaikan tidak akan dijalankan ulang. Flag ini berguna terutama saat Anda menguji pipeline karena saat caching diaktifkan, operasi akan selesai lebih cepat dan menggunakan lebih sedikit resource.

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()

Mengotomatiskan pipeline

Pada tahap ini, Anda telah meluncurkan pipeline pertama. Anda dapat memeriksa halaman Vertex AI Pipelines di konsol untuk melihat status tugas ini. Anda dapat melihat setiap container dibuat dan dijalankan. Anda juga dapat melacak error untuk komponen tertentu pada bagian ini dengan mengklik setiap komponen.

Untuk menjadwalkan pipeline, bangun Cloud Function dan gunakan penjadwal yang mirip dengan cron job.

Kode di bagian terakhir notebook menjadwalkan pipeline untuk dijalankan sekali sehari, seperti yang ditunjukkan dalam cuplikan kode berikut:

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
)

Menggunakan pipeline yang sudah selesai dalam produksi

Pipeline yang telah selesai telah melakukan tugas-tugas berikut:

• Membuat set data input.
• Melatih beberapa model menggunakan BigQuery ML serta XGBoost Python.
• Menganalisis hasil model.
• Men-deploy model XGBoost.

Anda juga telah mengotomatiskan pipeline dengan menggunakan Cloud Functions dan Cloud Scheduler untuk berjalan setiap hari.

Pipeline yang ditentukan di notebook dibuat untuk mengilustrasikan cara pembuatan berbagai model. Anda tidak akan menjalankan pipeline karena pipeline saat ini dibangun dalam skenario produksi. Namun, Anda dapat menggunakan pipeline ini sebagai panduan dan memodifikasi komponen sesuai kebutuhan. Misalnya, Anda dapat mengedit proses pembuatan fitur untuk memanfaatkan data Anda, mengubah rentang tanggal, dan mungkin membangun model alternatif. Anda juga akan memilih model dari beberapa ilustrasi yang paling sesuai dengan persyaratan produksi Anda.

Saat pipeline siap untuk produksi, Anda dapat mengimplementasikan tugas tambahan. Misalnya, Anda dapat menerapkan model pemenang/penantang, yang setiap hari membuat model baru dan model baru (penantang) dan yang sudah ada (juara) diberi skor berdasarkan data baru. Anda dapat memasukkan model baru ke dalam produksi hanya jika performanya lebih baik daripada performa model saat ini. Untuk memantau progres sistem, Anda juga dapat menyimpan catatan performa model setiap hari dan memvisualisasikan performa yang sedang tren.