Building a Custom BigQuery Data Pipeline for Scaleo and Affise Affiliate Data Integration

1. Context and Background

Modern digital marketing relies heavily on accurate data to evaluate channel performance and optimize spending. While primary advertising platforms like Google Ads or Meta provide native Data Transfer Services (DTS) to load data directly into Google BigQuery, affiliate marketing networks often lack these seamless, out-of-the-box integrations. Platforms like Scaleo and Affise are powerful tools for managing partner networks, tracking postbacks, and processing payouts. However, leaving this data isolated within their respective dashboards creates significant data silos.

When affiliate conversion data is isolated, data engineering and analytics teams struggle to build a complete map of the user journey. The lack of a centralized data warehouse means that evaluating the true Return on Ad Spend (ROAS) becomes a manual, error-prone process involving CSV exports and spreadsheet merges. To resolve this, organizations must build custom API integrations to extract raw data, load it into a unified data environment, and transform it for downstream analytics. Google BigQuery, combined with Google Cloud Run for serverless compute, provides a highly scalable and cost-effective infrastructure for this exact data engineering challenge.

2. Objective: Enriching Campaign Attribution

The primary objective of this architecture is the enrichment of the existing campaign attribution model by integrating external ad network data from Scaleo and Affise. By capturing granular conversion events—such as clicks, transaction IDs, postbacks, statuses, and payouts—and unifying them into a standardized schema, we can map external affiliate performance directly to internal session data.

This integration process ensures that every conversion tracked by a third-party affiliate network is accurately matched with the initial click or session stored in your core marketing attribution tables. The resulting pipeline will validate the architecture of the data flow, ensuring high data quality, strict deduplication, and automated injection into the main attribution model. Furthermore, building this entirely on Google Cloud Platform allows for strict FinOps optimization, ensuring that both the compute resources and the BigQuery storage and querying costs remain predictable and minimal.

3. Core Process: Python Connector Development

We will develop a robust, production-ready Python service utilizing the FastAPI framework. This application will be containerized using Docker and deployed on Google Cloud Run. It will expose specific HTTP endpoints (/fetch/scaleo and /fetch/affise) that a job scheduler can trigger.

The application will extract data via the REST APIs of both platforms, apply initial validation, and stream the raw JSON payloads into BigQuery flat tables.

3.1. Project Directory Structure Before writing the code, set up your local development environment with the following directory structure:

Plaintext

/affiliate_connector_project
 ├── app/
 │   ├── __init__.py
 │   ├── main.py
 │   ├── config.py
 │   ├── bq_client.py
 │   ├── scaleo_extractor.py
 │   └── affise_extractor.py
 ├── requirements.txt
 └── Dockerfile

3.2. Dependencies (requirements.txt) We require libraries for the web server, HTTP requests, Google Cloud interaction, and data validation.

Plaintext

fastapi==0.103.1
uvicorn==0.23.2
google-cloud-bigquery==3.11.4
requests==2.31.0
pydantic==2.3.0
tenacity==8.2.3

3.3. Configuration Management (app/config.py) Centralizing configuration ensures security and easier deployment. We utilize environment variables to handle sensitive API keys.

Python

import os
from pydantic import BaseModel

class Settings(BaseModel):
    GCP_PROJECT_ID: str = os.environ.get("GCP_PROJECT_ID", "default-project")
    BQ_DATASET: str = os.environ.get("BQ_DATASET", "raw_affiliate_data")
    
    SCALEO_API_KEY: str = os.environ.get("SCALEO_API_KEY", "")
    SCALEO_NETWORK_URL: str = os.environ.get("SCALEO_NETWORK_URL", "")
    
    AFFISE_API_KEY: str = os.environ.get("AFFISE_API_KEY", "")
    AFFISE_NETWORK_URL: str = os.environ.get("AFFISE_NETWORK_URL", "")

settings = Settings()

3.4. BigQuery Loader Module (app/bq_client.py) This module handles the interaction with BigQuery. We store the complete API response as a single JSON string. This approach (Extract and Load, then Transform) protects the pipeline from schema changes on the API side.

Python

from google.cloud import bigquery
import json
import logging
from app.config import settings

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
client = bigquery.Client(project=settings.GCP_PROJECT_ID)

def load_data_to_bq(table_name: str, raw_data: list) -> int:
    """
    Loads a list of dictionary records into a BigQuery table as raw JSON payloads.
    """
    if not raw_data:
        logging.warning(f"No data provided to load into {table_name}")
        return 0

    table_id = f"{settings.GCP_PROJECT_ID}.{settings.BQ_DATASET}.{table_name}"
    
    # Prepare data for insertion. We use 'payload' for the raw JSON and an ingestion timestamp.
    rows_to_insert = [
        {
            "payload": json.dumps(record),
            "ingestion_timestamp": "AUTO" # BigQuery will populate this via default value
        } 
        for record in raw_data
    ]
    
    job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("payload", "JSON", description="Raw JSON data from the API"),
            bigquery.SchemaField("ingestion_timestamp", "TIMESTAMP", default_value_expression="CURRENT_TIMESTAMP()", description="Time of ingestion")
        ],
        write_disposition="WRITE_APPEND",
    )
    
    try:
        job = client.load_table_from_json(rows_to_insert, table_id, job_config=job_config)
        job.result()  # Wait for the job to complete
        logging.info(f"Successfully loaded {job.output_rows} rows into {table_id}")
        return job.output_rows
    except Exception as e:
        logging.error(f"Failed to load data to BigQuery: {str(e)}")
        raise e

3.5. Scaleo API Extractor (app/scaleo_extractor.py) The Scaleo API requires specific headers and pagination handling. We use the tenacity library to implement retry logic for network resilience, which is a critical part of robust cloud architecture validation.

Python

import requests
import logging
from datetime import datetime, timedelta
from tenacity import retry, stop_after_attempt, wait_exponential
from app.config import settings
from app.bq_client import load_data_to_bq

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_scaleo_page(url: str, headers: dict, params: dict):
    response = requests.get(url, headers=headers, params=params, timeout=30)
    response.raise_for_status()
    return response.json()

def process_scaleo_data(date_from: str, date_to: str) -> dict:
    url = f"{settings.SCALEO_NETWORK_URL}/api/v2/network/statistics/conversions"
    headers = {"api-key": settings.SCALEO_API_KEY}
    
    page = 1
    per_page = 500
    all_conversions = []
    
    logging.info(f"Starting Scaleo extraction from {date_from} to {date_to}")
    
    while True:
        params = {
            "date_from": date_from,
            "date_to": date_to,
            "per_page": per_page,
            "page": page
        }
        
        try:
            data = fetch_scaleo_page(url, headers, params)
        except Exception as e:
            logging.error(f"Scaleo API failed after retries: {str(e)}")
            raise

        if data.get("code") != 200:
            error_msg = data.get("message", "Unknown Scaleo Error")
            logging.error(f"Scaleo API returned non-200 code: {error_msg}")
            raise ValueError(error_msg)
            
        conversions = data.get("info", {}).get("conversions", [])
        if not conversions:
            break
            
        all_conversions.extend(conversions)
        
        pagination = data.get("info", {}).get("pagination", {})
        page_count = pagination.get("page_count", 1)
        
        if page >= page_count:
            break
            
        page += 1
        
    rows_inserted = load_data_to_bq("raw_scaleo_conversions", all_conversions)
    return {"platform": "Scaleo", "status": "success", "rows_fetched": len(all_conversions), "rows_inserted": rows_inserted}

3.6. Affise API Extractor (app/affise_extractor.py) Affise operates similarly but has a different pagination structure and response schema. We must adapt the extraction logic accordingly.

Python

import requests
import logging
from datetime import datetime, timedelta
from tenacity import retry, stop_after_attempt, wait_exponential
from app.config import settings
from app.bq_client import load_data_to_bq

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_affise_page(url: str, headers: dict, params: dict):
    response = requests.get(url, headers=headers, params=params, timeout=30)
    response.raise_for_status()
    return response.json()

def process_affise_data(date_from: str, date_to: str) -> dict:
    url = f"{settings.AFFISE_NETWORK_URL}/3.0/stats/conversions"
    headers = {"API-Key": settings.AFFISE_API_KEY}
    
    page = 1
    limit = 500
    all_conversions = []
    
    logging.info(f"Starting Affise extraction from {date_from} to {date_to}")
    
    while True:
        params = {
            "date_from": date_from,
            "date_to": date_to,
            "limit": limit,
            "page": page
        }
        
        try:
            data = fetch_affise_page(url, headers, params)
        except Exception as e:
            logging.error(f"Affise API failed after retries: {str(e)}")
            raise

        if data.get("status") != 1:
            error_msg = data.get("message", "Unknown Affise Error")
            logging.error(f"Affise API error: {error_msg}")
            raise ValueError(error_msg)
            
        conversions = data.get("conversions", [])
        if not conversions:
            break
            
        all_conversions.extend(conversions)
        
        pagination = data.get("pagination", {})
        total_count = pagination.get("total_count", 0)
        
        if page * limit >= total_count:
            break
            
        page += 1
        
    rows_inserted = load_data_to_bq("raw_affise_conversions", all_conversions)
    return {"platform": "Affise", "status": "success", "rows_fetched": len(all_conversions), "rows_inserted": rows_inserted}

3.7. FastAPI Web Application (app/main.py) This script binds the extractors to HTTP POST endpoints. It automatically calculates the date range (yesterday) to fetch data for the most recently completed day.

Python

from fastapi import FastAPI, HTTPException
from datetime import datetime, timedelta
from app.scaleo_extractor import process_scaleo_data
from app.affise_extractor import process_affise_data
import logging

app = FastAPI(title="Affiliate Network Data Connector")

def get_yesterday_dates():
    yesterday = datetime.utcnow() - timedelta(days=1)
    date_str = yesterday.strftime('%Y-%m-%d')
    return date_str, date_str

@app.post("/trigger/scaleo")
def trigger_scaleo():
    date_from, date_to = get_yesterday_dates()
    try:
        result = process_scaleo_data(date_from, date_to)
        return result
    except Exception as e:
        logging.error(f"Scaleo pipeline failed: {str(e)}")
        raise HTTPException(status_code=500, detail="Scaleo extraction process failed.")

@app.post("/trigger/affise")
def trigger_affise():
    date_from, date_to = get_yesterday_dates()
    try:
        result = process_affise_data(date_from, date_to)
        return result
    except Exception as e:
        logging.error(f"Affise pipeline failed: {str(e)}")
        raise HTTPException(status_code=500, detail="Affise extraction process failed.")

@app.get("/health")
def health_check():
    return {"status": "healthy"}

3.8. Dockerizing the Application (Dockerfile) To deploy on Google Cloud Run, the application must be packaged into a Docker container. We use a slim Python image to minimize the attack surface and reduce image size.

Dockerfile

# Use the official lightweight Python image
FROM python:3.10-slim

# Set the working directory inside the container
WORKDIR /app

# Copy the requirements file and install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy the application code
COPY ./app /app/app

# Set environment variables for Python behavior
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Expose the port Cloud Run expects
EXPOSE 8080

# Command to run the uvicorn server
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

4. Deployment on Google Cloud Run

Google Cloud Run provides a fully managed serverless execution environment. It automatically scales to zero when not in use, adhering to strict FinOps principles for cost optimization.

4.1. Build and Deploy Process Ensure you have the Google Cloud CLI (gcloud) installed and authenticated. Open your terminal in the root directory of the project.

Bash

# Define variables
PROJECT_ID="your-gcp-project-id"
REGION="europe-west1"
SERVICE_NAME="affiliate-data-connector"
IMAGE_URI="gcr.io/$PROJECT_ID/$SERVICE_NAME:latest"
SERVICE_ACCOUNT="data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com"

# 1. Enable necessary Google Cloud APIs
gcloud services enable run.googleapis.com \
    cloudbuild.googleapis.com \
    cloudscheduler.googleapis.com \
    bigquery.googleapis.com

# 2. Build the Docker image using Google Cloud Build
gcloud builds submit --tag $IMAGE_URI

# 3. Deploy the container to Google Cloud Run
gcloud run deploy $SERVICE_NAME \
  --image $IMAGE_URI \
  --platform managed \
  --region $REGION \
  --no-allow-unauthenticated \
  --service-account $SERVICE_ACCOUNT \
  --set-env-vars GCP_PROJECT_ID=$PROJECT_ID,BQ_DATASET="raw_affiliate_data",SCALEO_API_KEY="your_key",SCALEO_NETWORK_URL="https://yournetwork.scaleo.io",AFFISE_API_KEY="your_key",AFFISE_NETWORK_URL="https://api.affise.com"

4.2. Configuring Google Cloud Scheduler To automate the data extraction, we will configure Cloud Scheduler to trigger our Cloud Run endpoints daily at 02:00 AM UTC.

Bash

# Retrieve the deployed Cloud Run service URL
SERVICE_URL=$(gcloud run services describe $SERVICE_NAME --platform managed --region $REGION --format 'value(status.url)')

# Create a scheduled job for Scaleo
gcloud scheduler jobs create http trigger-scaleo-daily \
  --schedule="0 2 * * *" \
  --uri="$SERVICE_URL/trigger/scaleo" \
  --http-method=POST \
  --oidc-service-account-email=$SERVICE_ACCOUNT \
  --oidc-token-audience=$SERVICE_URL \
  --location=$REGION

# Create a scheduled job for Affise
gcloud scheduler jobs create http trigger-affise-daily \
  --schedule="15 2 * * *" \
  --uri="$SERVICE_URL/trigger/affise" \
  --http-method=POST \
  --oidc-service-account-email=$SERVICE_ACCOUNT \
  --oidc-token-audience=$SERVICE_URL \
  --location=$REGION

5. Dataform SQL Pipeline: Unpacking, Quality, Deduplication, and Injection

Dataform allows us to manage complex SQL workflows in BigQuery as code. The following setup will transform the raw JSON data into a clean, structured format and inject it into the marketing attribution model.

5.1. Dataform Project Initialization In your Dataform repository, ensure your dataform.json points to the correct default BigQuery location and default schema.

dataform.json

JSON

{
  "defaultSchema": "affiliate_transformations",
  "assertionSchema": "dataform_assertions",
  "warehouse": "bigquery",
  "defaultDatabase": "your-gcp-project-id",
  "defaultLocation": "EU"
}

5.2. Source Declarations Create a file named definitions/sources.js to declare the raw tables created by the Python connector.

JavaScript

declare({
  database: "your-gcp-project-id",
  schema: "raw_affiliate_data",
  name: "raw_scaleo_conversions",
  description: "Raw JSON payloads from the Scaleo API"
});

declare({
  database: "your-gcp-project-id",
  schema: "raw_affiliate_data",
  name: "raw_affise_conversions",
  description: "Raw JSON payloads from the Affise API"
});

5.3. Unpacking and Deduplicating Scaleo Data Create definitions/staging/stg_scaleo_conversions.sqlx. This script extracts fields from the JSON payload and handles data deduplication. Because the Python connector uses an “append-only” strategy, duplicates may occur if the job runs multiple times. We resolve this by picking the row with the latest ingestion_timestamp.

SQL

config {
  type: "incremental",
  schema: "staging",
  uniqueKey: ["conversion_id"],
  bigquery: {
    partitionBy: "DATE(conversion_timestamp)",
    clusterBy: ["offer_id", "status"]
  },
  description: "Flattened and deduplicated conversions from Scaleo",
  assertions: {
    nonNull: ["conversion_id", "click_id", "conversion_timestamp"]
  }
}

WITH extracted_data AS (
  SELECT
    JSON_VALUE(payload, '$.id') AS conversion_id,
    JSON_VALUE(payload, '$.click_id') AS click_id,
    JSON_VALUE(payload, '$.offer.id') AS offer_id,
    JSON_VALUE(payload, '$.offer.name') AS offer_name,
    JSON_VALUE(payload, '$.affiliate.id') AS affiliate_id,
    CAST(JSON_VALUE(payload, '$.payout') AS FLOAT64) AS payout,
    CAST(JSON_VALUE(payload, '$.revenue') AS FLOAT64) AS revenue,
    LOWER(JSON_VALUE(payload, '$.status')) AS status,
    CAST(JSON_VALUE(payload, '$.added_timestamp') AS TIMESTAMP) AS conversion_timestamp,
    ingestion_timestamp
  FROM
    ${ref("raw_scaleo_conversions")}
  ${when(incremental(), `WHERE ingestion_timestamp > (SELECT MAX(ingestion_timestamp) FROM ${self()})`)}
),

ranked_data AS (
  SELECT
    *,
    ROW_NUMBER() OVER(PARTITION BY conversion_id ORDER BY ingestion_timestamp DESC) as row_num
  FROM extracted_data
)

SELECT
  conversion_id,
  click_id,
  offer_id,
  offer_name,
  affiliate_id,
  payout,
  revenue,
  status,
  conversion_timestamp,
  ingestion_timestamp
FROM ranked_data
WHERE row_num = 1

5.4. Unpacking and Deduplicating Affise Data Create definitions/staging/stg_affise_conversions.sqlx. Note the slight differences in the JSON path extraction corresponding to Affise’s data structure.

SQL

config {
  type: "incremental",
  schema: "staging",
  uniqueKey: ["conversion_id"],
  bigquery: {
    partitionBy: "DATE(conversion_timestamp)",
    clusterBy: ["offer_id", "status"]
  },
  description: "Flattened and deduplicated conversions from Affise",
  assertions: {
    nonNull: ["conversion_id", "click_id", "conversion_timestamp"]
  }
}

WITH extracted_data AS (
  SELECT
    JSON_VALUE(payload, '$.id') AS conversion_id,
    JSON_VALUE(payload, '$.click_id') AS click_id,
    JSON_VALUE(payload, '$.offer_id') AS offer_id,
    JSON_VALUE(payload, '$.offer_title') AS offer_name,
    JSON_VALUE(payload, '$.partner_id') AS affiliate_id,
    CAST(JSON_VALUE(payload, '$.sum') AS FLOAT64) AS payout,
    CAST(JSON_VALUE(payload, '$.revenue') AS FLOAT64) AS revenue,
    LOWER(JSON_VALUE(payload, '$.status')) AS status,
    CAST(JSON_VALUE(payload, '$.created_at') AS TIMESTAMP) AS conversion_timestamp,
    ingestion_timestamp
  FROM
    ${ref("raw_affise_conversions")}
  ${when(incremental(), `WHERE ingestion_timestamp > (SELECT MAX(ingestion_timestamp) FROM ${self()})`)}
),

ranked_data AS (
  SELECT
    *,
    ROW_NUMBER() OVER(PARTITION BY conversion_id ORDER BY ingestion_timestamp DESC) as row_num
  FROM extracted_data
)

SELECT
  conversion_id,
  click_id,
  offer_id,
  offer_name,
  affiliate_id,
  payout,
  revenue,
  status,
  conversion_timestamp,
  ingestion_timestamp
FROM ranked_data
WHERE row_num = 1

5.5. Unifying the Schema for Downstream Use Create definitions/intermediate/int_unified_affiliate_conversions.sqlx. This view merges both network datasets into a single, standardized table, filtering only for valid, paid conversions.

SQL

config {
  type: "view",
  schema: "intermediate",
  description: "Unified view of approved conversions from all external affiliate networks."
}

SELECT
  'Scaleo' AS network_source,
  conversion_id,
  click_id,
  offer_id,
  offer_name,
  affiliate_id,
  payout AS affiliate_cost,
  revenue AS affiliate_revenue,
  status,
  conversion_timestamp
FROM ${ref("stg_scaleo_conversions")}
WHERE status = 'approved'

UNION ALL

SELECT
  'Affise' AS network_source,
  conversion_id,
  click_id,
  offer_id,
  offer_name,
  affiliate_id,
  payout AS affiliate_cost,
  revenue AS affiliate_revenue,
  status,
  conversion_timestamp
FROM ${ref("stg_affise_conversions")}
WHERE status = 'confirmed' -- Affise uses 'confirmed' instead of 'approved'

5.6. Injecting into the Main Attribution Model Finally, create definitions/marts/fct_marketing_attribution_enriched.sqlx. This table joins your existing core marketing clicks (generated by your internal tracking system or server-side Google Tag Manager) with the unified affiliate conversions based on the unique click_id.

SQL

config {
  type: "table",
  schema: "marts",
  description: "Final enriched attribution model combining internal clicks with external network conversions.",
  bigquery: {
    partitionBy: "DATE(click_timestamp)",
    clusterBy: ["utm_source", "utm_medium"]
  }
}

WITH base_internal_clicks AS (
  -- Assume this table exists in your data warehouse, populated by your internal tracker
  SELECT
    session_id,
    click_id,
    utm_source,
    utm_medium,
    utm_campaign,
    device_category,
    click_timestamp
  FROM `your-gcp-project-id.core_marketing.fct_clicks`
),

external_conversions AS (
  SELECT
    network_source,
    click_id,
    offer_name,
    affiliate_cost,
    affiliate_revenue,
    conversion_timestamp
  FROM ${ref("int_unified_affiliate_conversions")}
)

SELECT
  c.session_id,
  c.click_id,
  c.utm_source,
  c.utm_medium,
  c.utm_campaign,
  c.device_category,
  c.click_timestamp,
  e.network_source,
  e.offer_name,
  COALESCE(e.affiliate_cost, 0) AS affiliate_cost,
  COALESCE(e.affiliate_revenue, 0) AS affiliate_revenue,
  e.conversion_timestamp,
  CASE WHEN e.click_id IS NOT NULL THEN TRUE ELSE FALSE END AS is_converted,
  TIMESTAMP_DIFF(e.conversion_timestamp, c.click_timestamp, MINUTE) as time_to_convert_minutes
FROM base_internal_clicks c
LEFT JOIN external_conversions e
  ON c.click_id = e.click_id

Синтез

The technical implementation described above establishes a resilient, automated data pipeline. However, deploying enterprise-grade data architectures requires synthesizing a few advanced strategies to ensure long-term stability, strict security, and cost efficiency.

1. Managing Conversion Status Updates (Lookback Windows) The default Cloud Scheduler configuration fetches data for the previous day (yesterday). In affiliate marketing, it is common practice for conversion statuses to change days or even weeks after the initial transaction (e.g., a transaction moves from “pending” to “approved”, or is “declined” due to fraud). If your pipeline only requests yesterday’s data, you will miss these historical status updates. To resolve this, you should synthesize a dynamic lookback window logic in the Python script. Instead of fetching exactly T-1, modify the Cloud Scheduler to run a secondary job weekly that fetches T-14 to T-0. Because the Dataform SQL configuration utilizes an incremental strategy coupled with a ROW_NUMBER() OVER(PARTITION BY conversion_id ORDER BY ingestion_timestamp DESC) deduplication logic, BigQuery will seamlessly overwrite the old “pending” record with the new “approved” record without generating duplicates in your final attribution model.

2. FinOps and BigQuery Optimization Strategy When designing data engineering pipelines on Google Cloud Platform, telemetry-driven performance optimization is critical to control FinOps metrics. In the Dataform SQLX files provided, strict partitioning (partitionBy: "DATE(conversion_timestamp)") and clustering (clusterBy: ["offer_id", "status"]) rules were applied to the tables. This is not arbitrary. When BI tools (like Looker) or data scientists query the fct_marketing_attribution_enriched table, they almost always filter by date ranges. Table partitioning limits the volume of data scanned by BigQuery during these queries, which directly reduces the financial cost per query. Without partitioning, every query would scan the entire historical dataset, leading to exponential cost increases as the dataset grows.

3. Infrastructure Security and Secret Management While passing API keys as environment variables in the gcloud run deploy command is functional for testing, it is not compliant with strict enterprise security standards. For a production deployment, you must integrate Google Cloud Secret Manager. You store the SCALEO_API_KEY and AFFISE_API_KEY in Secret Manager and map these secrets to the Cloud Run service instance. This ensures that the secrets are injected directly into the container’s memory at runtime and are never exposed in plaintext within CI/CD logs, Terraform state files, or the Cloud Run console configuration.

4. Advanced Error Monitoring and Architecture Validation Deploying to Cloud Run is only the first step; continuous architecture validation is required. You should configure Google Cloud Operations Suite (formerly Stackdriver) to monitor the execution of the Cloud Run instances. Set up Log-Based Metrics to track the frequency of error logs generated by the Python application (e.g., when the tenacity retry block fails entirely). Create an alerting policy that sends a notification to a Pub/Sub topic or a Slack channel if the error rate exceeds a defined threshold. This proactive telemetry ensures that any disruptions in the affiliate APIs or network connectivity are identified and resolved before they cause significant gaps in your main marketing attribution model.