Event-Driven ML Workflow Execution with GCP Workflows

Question

Pulumi · Accepted Answer

Event-driven workflows in machine learning on Google Cloud Platform (GCP) typically involve triggering a series of tasks in response to certain events — such as data arriving in a storage bucket. GCP Workflows is a managed orchestration service that allows you to sequence and execute tasks defined in a serverless manner. Combined with Cloud Functions, Pub/Sub, and various AI and machine learning services, you can create robust event-driven ML workflows.

Let's go through setting up an event-driven ML workflow using GCP Workflows:

1. **Cloud Storage Bucket**: This is where your data (e.g., new images for processing) will be uploaded. An event is triggered when new data is added.
   
2. **Pub/Sub**: This messaging service will publish messages to notify subscribers that a new file has been uploaded.

3. **Cloud Function**: This listens for messages from the Pub/Sub topic and kickstarts the GCP Workflows execution upon receiving a message, which could include preprocessing the data and sending it to AI Platform for prediction.

4. **GCP Workflows**: You define a serverless workflow here, which might include various steps such as invoking AI Platform models for predictions and storing the results, as well as any post-processing tasks.

5. **AI Platform**: This is where your machine learning model resides. It will receive data for inference, process it, and return predictions.

Below is a program written in Python using Pulumi to define a simple event-driven ML workflow on GCP:

```python
import pulumi
import pulumi_gcp as gcp

# Replace these variables with your specific settings
project = "your-gcp-project"
region = "us-central1"
ai_platform_model_name = "your_model"

# Create a Cloud Storage bucket where data will be uploaded.
bucket = gcp.storage.Bucket("bucket", name="your-bucket-name")

# Create a Pub/Sub topic that will notify of new data in the bucket.
topic = gcp.pubsub.Topic("topic", name="data-uploaded-topic")

# Set the bucket notification for the Pub/Sub topic
# so that every time a new object is created, a message will be sent to the topic.
bucket_notification = gcp.storage.Notification("bucketNotification",
    bucket=bucket.name,
    payload_format="JSON_API_V1",
    topic=topic.name
)

# Define a simple Cloud Function that starts a GCP workflow on Pub/Sub events.
function = gcp.cloudfunctions.Function("function",
    name="start-workflow",
    entry_point="trigger_workflow",
    runtime="python37",
    available_memory_mb=128,
    source_archive_bucket=bucket.name,
    trigger_http=True,
    event_trigger=gcp.cloudfunctions.FunctionEventTriggerArgs(
        event_type="google.pubsub.topic.publish",
        resource=topic.id
    ),
)

# Define a GCP Workflow that is triggered by Cloud Function and calls the AI Platform for predictions.
workflow = gcp.workflows.Workflow("workflow",
    name="ml-prediction-workflow",
    region=region,
    service_account=gcp.serviceAccount.Account("workflow-sa").email,
    source_contents=f"""
- init:
    call: http.post
    args:
        url: https://{region}-aiplatform.googleapis.com/v1/projects/{project}/locations/{region}/models/{ai_platform_model_name}:predict
        body:
            instances: ${{{"{"}}body.instances}}   # Data passed from the triggering event
    result: prediction
- returnResult:
    return: ${{{"{"}}prediction}}
"""
)

# Export useful information
pulumi.export('bucket_name', bucket.name)
pulumi.export('pub_sub_topic', topic.name)
pulumi.export('cloud_function_name', function.name)
pulumi.export('workflow_name', workflow.name)
```

### Explanation
- First, we create a **Cloud Storage bucket** (`gcp.storage.Bucket`), which serves as the location where the data files triggering the event-driven workflow will be uploaded.
  
- Next, we set up a **Pub/Sub topic** (`gcp.pubsub.Topic`) that will publish messages when new data is added to the bucket.

- The `bucket_notification` resource sets up a **notification** on the bucket to send messages to the Pub/Sub topic we created.

- A **Cloud Function** (`gcp.cloudfunctions.Function`) listens for these messages and begins the workflow execution defined in GCP Workflows.

- The **GCP Workflow** (`gcp.workflows.Workflow`) defines a workflow that will be started by the Cloud Function and makes a call to the AI Platform to get predictions for the new data.

- Finally, we export the names of the resources created as stack outputs. This information can be used for reference or integration with other systems.

### Things to Note
- This is a simplified example to illustrate the setup of an event-driven ML workflow using GCP Workflows; a real-world scenario would have additional security measures, error handling, and possibly multiple workflow steps.
  
- The actual logic within the Cloud Function (`entry_point="trigger_workflow"`) is not provided in the code, you will need to define the Python handler function that would extract event details and start the workflow.

- The `source_contents` of the workflow is containing only a simple HTTP Post call as a placeholder. In a real scenario, you would flesh this out with appropriate initialization and the needed arguments to call your AI Platform model.

- This workflow assumes that the AI Platform model endpoint and the appropriate request body schema are known. You must replace `"your_model"` with the name of your actual model.

- You need to replace `"your-gcp-project"`, `"your-bucket-name"`, and other placeholder values with appropriate values for your setup.

Remember to package the source code for your Cloud Function and upload it to the specified Cloud Storage bucket before deploying this configuration. Also, ensure your GCP Workflows service account (`workflow-sa`) has sufficient permissions for the required operations.