AutoMLOpsを使って機械学習CI/CDパイプラインを組んでみた

.
├── .env   # リージョン指定などに使用する環境変数
├── data   # BQ にアップロードしたデータセット
│   └── Dry_Beans_Dataset.csv
└── src
    ├── callable_functions.py   # component を用意
    ├── main.py   # pipeline を組み, AutoMLOps を実行
    ├── inference.py   # 推論用のコードを記述
    └── upload_to_bq.py   # 事前準備で使用

from google_cloud_automlops import AutoMLOps

@AutoMLOps.component(
    packages_to_install=[
        'google-cloud-bigquery', 
        'pandas',
        'pyarrow',
        'db_dtypes',
        'fsspec',
        'gcsfs'
    ]
)
def create_dataset(
    bq_table: str,
    data_path: str,
    project_id: str
):
    from google.cloud import bigquery
    import pandas as pd
    from sklearn import preprocessing

    bq_client = bigquery.Client(project=project_id)

    def get_query(bq_input_table: str) -> str:
        # sample code そのままですが、実行時はご注意ください
        return f'''
        SELECT *
        FROM {bq_input_table}
        '''

    def load_bq_data(query: str, client: bigquery.Client) -> pd.DataFrame:
        df = client.query(query).to_dataframe()
        return df

    dataframe = load_bq_data(get_query(bq_table), bq_client)
    le = preprocessing.LabelEncoder()
    dataframe['Class'] = le.fit_transform(dataframe['Class'])
    dataframe.to_csv(data_path, index=False)

@AutoMLOps.component(
    packages_to_install=[
        'scikit-learn==1.2.2',
        'pandas',
        'joblib',
        'tensorflow'
    ]
)
def train_model(
    data_path: str,
    model_directory: str
):
    from sklearn.tree import DecisionTreeClassifier
    from sklearn.model_selection import train_test_split
    import pandas as pd
    import tensorflow as tf
    import pickle
    import os

    def save_model(model, uri):
        """Saves a model to uri."""
        with tf.io.gfile.GFile(uri, 'w') as f:
            pickle.dump(model, f)

    df = pd.read_csv(data_path)
    labels = df.pop('Class').tolist()
    data = df.values.tolist()
    x_train, x_test, y_train, y_test = train_test_split(data, labels)
    skmodel = DecisionTreeClassifier()
    skmodel.fit(x_train,y_train)
    score = skmodel.score(x_test,y_test)
    print('accuracy is:',score)

    output_uri = os.path.join(model_directory, 'model.pkl')
    save_model(skmodel, output_uri)

@AutoMLOps.component(
    packages_to_install=[
        'google-cloud-aiplatform'
    ]
)
def deploy_model(
    model_directory: str,
    project_id: str,
    region: str
):
    import pprint as pp
    import random

    from google.cloud import aiplatform

    aiplatform.init(project=project_id, location=region)
    # Check if model exists
    models = aiplatform.Model.list()
    model_name = 'beans-model'
    if 'beans-model' in (m.name for m in models):
        parent_model = model_name
        model_id = None
        is_default_version=False
        version_aliases=['experimental', 'challenger', 'custom-training', 'decision-tree']
        version_description='challenger version'
    else:
        parent_model = None
        model_id = model_name
        is_default_version=True
        version_aliases=['champion', 'custom-training', 'decision-tree']
        version_description='first version'

    serving_container = 'us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-2:latest'
    uploaded_model = aiplatform.Model.upload(
        artifact_uri=model_directory,
        model_id=model_id,
        display_name=model_name,
        parent_model=parent_model,
        is_default_version=is_default_version,
        version_aliases=version_aliases,
        version_description=version_description,
        serving_container_image_uri=serving_container,
        serving_container_ports=[8080],
        labels={'created_by': 'automlops-team'},
    )

    endpoint = uploaded_model.deploy(
        machine_type='n1-standard-4',
        deployed_model_display_name='deployed-beans-model')

    sample_input = [[random.uniform(0, 300) for x in range(16)]]

    # Test endpoint predictions
    print('running prediction test...')
    try:
        resp = endpoint.predict(instances=sample_input)
        pp.pprint(resp)
    except Exception as ex:
        print('prediction request failed', ex)

import os
import sys
import datetime
import argparse
from dotenv import load_dotenv
from google.cloud import aiplatform
from google_cloud_automlops import AutoMLOps

from callable_functions import create_dataset, train_model, deploy_model
from inference import run_inference

load_dotenv()

def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--operation', type=str, required=True,
                          choices=['provision', 'deploy', 'generate', 'monitor'])
    return parser.parse_args()

@AutoMLOps.pipeline
def pipeline(
    bq_table: str,
    model_directory: str,
    data_path: str,
    project_id: str,
    region: str):

    create_dataset_task = create_dataset(
        bq_table=bq_table,
        data_path=data_path,
        project_id=project_id)

    train_model_task = train_model(
        model_directory=model_directory,
        data_path=data_path).after(create_dataset_task)

    deploy_model_task = deploy_model(
        model_directory=model_directory,
        project_id=project_id,
        region=region).after(train_model_task)

pipeline_params = {
    'bq_table': os.getenv('TRAINING_DATASET'),
    'model_directory': f'gs://{os.getenv("PROJECT_ID")}-{os.getenv("MODEL_ID")}-bucket/trained_models/{datetime.datetime.now()}',
    'data_path': f'gs://{os.getenv("PROJECT_ID")}-{os.getenv("MODEL_ID")}-bucket/data.csv',
    'project_id': os.getenv('PROJECT_ID'),
    'region': os.getenv('REGION'),
}

if __name__ == '__main__':
    args = get_args()

    if args.operation == 'generate':
        AutoMLOps.generate(
            project_id=os.getenv('PROJECT_ID'),
            pipeline_params=pipeline_params,
            use_ci=True,
            naming_prefix=os.getenv('MODEL_ID'),
            schedule_pattern='59 11 * * 0', # retrain every Sunday at Midnight
            setup_model_monitoring=True,   # use this if you would like to use Vertex Model Mointoring
            artifact_repo_location=os.getenv('REGION'),
            build_trigger_location=os.getenv('REGION'),
            pipeline_job_submission_service_location=os.getenv('REGION'),
            schedule_location=os.getenv('REGION'),
            storage_bucket_location=os.getenv('REGION'),
        )

    if args.operation == 'provision':
        AutoMLOps.provision(hide_warnings=False)

    if args.operation == 'deploy':
        AutoMLOps.deploy(precheck=True, hide_warnings=False)

    # optional
    if args.operation == 'monitor':
        aiplatform.init(project=os.getenv("PROJECT_ID"), location=os.getenv("REGION"))
        endpoints = aiplatform.Endpoint.list(filter='display_name="beans-model_endpoint"')
        print(aiplatform.Endpoint.list())
        if not endpoints:
            print('No endpoint found. Please deploy the model first.')
            sys.exit()
        else:
            AutoMLOps.monitor(
                alert_emails=['email'],
                target_field='Class',
                model_endpoint=aiplatform.Endpoint.list(filter='display_name="beans-model_endpoint"')[0].resource_name,
                monitoring_interval=1,
                auto_retraining_params=pipeline_params,
                drift_thresholds={'Area': 0.000001, 'Perimeter': 0.000001},
                skew_thresholds={'Area': 0.000001, 'Perimeter': 0.000001},
                training_dataset=f'bq://{os.getenv("TRAINING_DATASET")}',
                hide_warnings=False,
                monitoring_location=os.getenv('REGION'),
            )
            run_inference()

python src/main.py --operation generate

python src/main.py --operation provision

python src/main.py --operation deploy

import os
from dotenv import load_dotenv
from sklearn.preprocessing import LabelEncoder
from sklearn.metrics import classification_report
from google.cloud import bigquery, aiplatform
import pandas as pd

load_dotenv()

def get_query(bq_input_table: str) -> str:
    # sample code そのままですが、実行時はご注意ください
    return f'''
        SELECT * FROM {bq_input_table}
    '''

def load_bq_data(query: str, client: bigquery.Client) -> pd.DataFrame:
    df = client.query(query).to_dataframe()
    return df

def run_inference(num_samples: int=30):
    bq_client = bigquery.Client(project=os.getenv('PROJECT_ID'))
    df = load_bq_data(get_query(os.getenv('TRAINING_DATASET')), bq_client)
    sampled_df = df.sample(n=num_samples)
    X_sample = sampled_df.iloc[:, :-1][:num_samples].values.tolist()
    Y_sample = sampled_df.iloc[:, -1][:num_samples].values.tolist()

    endpoints = aiplatform.Endpoint.list(filter=f'display_name="beans-model_endpoint"')
    endpoint_name = endpoints[0].resource_name
    endpoint = aiplatform.Endpoint(endpoint_name)
    response = endpoint.predict(instances=X_sample)
    predictions = response[0]

    # calc accuracy
    le = LabelEncoder()
    y_true = df.iloc[:, -1].values.tolist()
    le.fit(y_true)
    y_true_encoded = le.transform(Y_sample)

    print(classification_report(y_true_encoded, predictions))

              precision    recall  f1-score   support

           0       1.00      1.00      1.00         3
           1       1.00      1.00      1.00         2
           2       1.00      1.00      1.00         3
           3       1.00      1.00      1.00        11
           4       1.00      1.00      1.00         5
           5       1.00      1.00      1.00         5
           6       1.00      1.00      1.00         1

    accuracy                           1.00        30
   macro avg       1.00      1.00      1.00        30
weighted avg       1.00      1.00      1.00        30