<jupyter_start><jupyter_text>Outline 1. Create Pub/Sub Topic ([refer](https://github.com/sayakpaul/CI-CD-for-Model-Training/blob/main/cloud_function_trigger.ipynb))2. Deploy Cloud Function ([refer](https://github.com/sayakpaul/CI-CD-for-Model-Training/blob/main/cloud_function_trigger.ipynb))    - check if there are enough number of images in a specific GCS bucket3. Publish Pub/Sub Topic to trigger batch prediction pipeline ([refer](https://github.com/sayakpaul/CI-CD-for-Model-Training/blob/main/cloud_scheduler_trigger.ipynb))    - need pipeline JSON spec somewhere in GCS  Setup<jupyter_code>!pip install --upgrade -q google-cloud-scheduler
!gcloud init
from google.colab import auth

auth.authenticate_user()
GOOGLE_CLOUD_PROJECT = "gcp-ml-172005"  # @param {type:"string"}
GOOGLE_CLOUD_REGION = "us-central1"

GCS_BUCKET_NAME = "cifar10-experimental-csp2"  # @param {type:"string"}
PIPELINE_NAME = "continuous-adaptation-for-data-changes-batch"  # @param {type:"string"}
PIPELINE_ROOT = "gs://{}/pipeline_root/{}".format(GCS_BUCKET_NAME, PIPELINE_NAME)
PIPELINE_LOCATION = f"{PIPELINE_ROOT}/{PIPELINE_NAME}_pipeline.json"

PUBSUB_TOPIC = f"trigger-{PIPELINE_NAME}"

SCHEDULER_JOB_NAME = f"scheduler-job-{PUBSUB_TOPIC}"

IMAGE_LOCATION_BUCKET = "batch-prediction-collection-3"  # @param {type:"string"}
IMAGE_LOCATION_BUCKET<jupyter_output><empty_output><jupyter_text>Create Pub/Sub Topic<jupyter_code>!gcloud pubsub topics create {PUBSUB_TOPIC}<jupyter_output>Created topic [projects/gcp-ml-172005/topics/trigger-continuous-adaptation-for-data-changes-batch].<jupyter_text>Deploy Cloud Function  Create Cloud Function Directory<jupyter_code>!mkdir -p cloud_function
!touch cloud_function/__init__.py<jupyter_output><empty_output><jupyter_text>Create Requirements.txt<jupyter_code>_cloud_function_dep = "cloud_function/requirements.txt"
%%writefile {_cloud_function_dep}

kfp==1.6.2
google-cloud-aiplatform
google-cloud-storage<jupyter_output>Writing cloud_function/requirements.txt<jupyter_text>Create Cloud Function Module<jupyter_code>!pip install google-cloud-storage
_cloud_function_file = "cloud_function/main.py"
%%writefile {_cloud_function_file}

import os
import re
import json
import logging
import base64

from datetime import datetime

from kfp.v2.google.client import AIPlatformClient
from google.cloud import storage


def get_number_of_images(storage_client, bucket, latest_directory):
    blobs = storage_client.list_blobs(bucket, prefix=latest_directory)

    count = 0
    for blob in blobs:
        if blob.name.split(".")[-1] == "jpg":
            count = count + 1

    return count


def is_there_enough_images(storage_client, bucket, latest_directory, threshold):
    number_of_images = get_number_of_images(storage_client, bucket, latest_directory)
    print(f"number of images = {number_of_images}")
    return number_of_images >= threshold


def get_latest_directory(storage_client, bucket):
    blobs = storage_client.list_blobs(bucket)

    folders = list(
        set(
            [
                os.path.dirname(blob.name)
                for blob in blobs
                if bool(
                    re.match(
                        "[1-9][0-9][0-9][0-9]-[0-1][0-9]", os.path.dirname(blob.name)
                    )
                )
                is True
            ]
        )
    )

    folders.sort(key=lambda date: datetime.strptime(date, "%Y-%m"))
    print(folders[0])
    return folders[0]


def trigger_pipeline(event, context):
    # Parse the environment variables.
    project = os.getenv("PROJECT")
    region = os.getenv("REGION")
    gcs_pipeline_file_location = os.getenv("GCS_PIPELINE_FILE_LOCATION")
    gcs_image_bucket = os.getenv("GCS_IMAGE_BUCKET")

    print(project)
    print(region)
    print(gcs_pipeline_file_location)
    print(gcs_image_bucket)

    threshold = 100

    # Check if the pipeline file exists in the provided GCS Bucket.
    storage_client = storage.Client()
    latest_directory = get_latest_directory(storage_client, gcs_image_bucket)

    if is_there_enough_images(
        storage_client, gcs_image_bucket, latest_directory, threshold
    ):
        path_parts = gcs_pipeline_file_location.replace("gs://", "").split("/")
        pipeline_bucket = path_parts[0]
        pipeline_blob = "/".join(path_parts[1:])

        pipeline_bucket = storage_client.bucket(pipeline_bucket)
        blob = storage.Blob(bucket=pipeline_bucket, name=pipeline_blob)

        if not blob.exists(storage_client):
            raise ValueError(f"{gcs_pipeline_file_location} does not exist.")

        # Initialize Vertex AI API client and submit for pipeline execution.
        api_client = AIPlatformClient(project_id=project, region=region)

        response = api_client.create_run_from_job_spec(
            job_spec_path=gcs_pipeline_file_location,
            parameter_values={"data_gcs_prefix": latest_directory},
            enable_caching=True,
        )

        logging.info(response)<jupyter_output>Overwriting cloud_function/main.py<jupyter_text>Deploy Cloud Function<jupyter_code>!cd cloud_function
ENV_VARS=f"""\
PROJECT={GOOGLE_CLOUD_PROJECT},\
REGION={GOOGLE_CLOUD_REGION},\
GCS_PIPELINE_FILE_LOCATION={PIPELINE_LOCATION},\
GCS_IMAGE_BUCKET={IMAGE_LOCATION_BUCKET}
"""

!echo {ENV_VARS}
BUCKET = f'gs://{GCS_BUCKET_NAME}'
CLOUD_FUNCTION_NAME = f'trigger-{PIPELINE_NAME}-fn'

!gcloud functions deploy {CLOUD_FUNCTION_NAME} \
    --region={GOOGLE_CLOUD_REGION} \
    --trigger-topic={PUBSUB_TOPIC} \
    --runtime=python37 \
    --source=cloud_function\
    --entry-point=trigger_pipeline\
    --stage-bucket={BUCKET}\
    --update-env-vars={ENV_VARS}<jupyter_output>For Cloud Build Logs, visit: https://console.cloud.google.com/cloud-build/builds;region=us-central1/b6851c4a-a1a5-47c2-8cb1-a22f927867ad?project=874401645461
availableMemoryMb: 256
buildId: 5dea4cb3-7602-4c7e-992a-2ddcaa5566bd
buildName: projects/874401645461/locations/us-central1/builds/5dea4cb3-7602-4c7e-992a-2ddcaa5566bd
entryPoint: trigger_pipeline
environmentVariables:
  GCS_IMAGE_BUCKET: batch-prediction-collection-3
  GCS_IMAGE_FILE_LOCATION: gs://batch-prediction-collection-3
  GCS_PIPELINE_FILE_LOCATION: gs://cifar10-experimental-csp2/pipeline_root/continuous-adaptation-for-data-changes-batch/continuous-adaptation-for-data-changes-batch_pipeline.json
  PROJECT: gcp-ml-172005
  REGION: us-central1
eventTrigger:
  eventType: google.pubsub.topic.publish
  failurePolicy: {}
  resource: projects/gcp-ml-172005/topics/trigger-continuous-adaptation-for-data-changes-batch
  service: pubsub.googleapis.com
ingressSettings: ALLOW_ALL
labels:
  deployment-tool: cli-gcloud
name: projects/g[...]<jupyter_text>See the Progress<jupyter_code>import IPython

cloud_fn_url = f"https://console.cloud.google.com/functions/details/{GOOGLE_CLOUD_REGION}/{CLOUD_FUNCTION_NAME}"
html = (
    f'See the Cloud Function details <a href="{cloud_fn_url}" target="_blank">here</a>.'
)
IPython.display.display(IPython.display.HTML(html))<jupyter_output><empty_output><jupyter_text>Create Cloud Scheduler's Job<jupyter_code>!gcloud scheduler jobs create pubsub $SCHEDULER_JOB_NAME --schedule "*/3 * * * *" --topic $PUBSUB_TOPIC --attributes name=scheduler #every hour<jupyter_output>name: projects/gcp-ml-172005/locations/us-central1/jobs/scheduler-job-trigger-continuous-adaptation-for-data-changes-batch
pubsubTarget:
  attributes:
    name: scheduler
  topicName: projects/gcp-ml-172005/topics/trigger-continuous-adaptation-for-data-changes-batch
retryConfig:
  maxBackoffDuration: 3600s
  maxDoublings: 16
  maxRetryDuration: 0s
  minBackoffDuration: 5s
schedule: '*/3 * * * *'
state: ENABLED
timeZone: Etc/UTC
userUpdateTime: '2021-10-18T01:10:04Z'
{'fraction_non_alphanumeric': 0.11232307492533437, 'fraction_numerical': 0.024412413972211403, 'mean_word_length': 4.61779722830051, 'pattern_counts': {'":': 2, '<': 30, '<?xml version=': 0, '>': 31, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '473134', 'license': 'Apache-2.0', 'n_tokens_mistral': 2777, 'n_tokens_neox': 2676, 'n_words': 403, 'path': 'notebooks/04_Cloud_Scheduler_Trigger.ipynb', 'repo_name': 'taocao/Continuous-Adaptation-for-Machine-Learning-System-to-Data-Changes', 'size': '7701'}