BigQuery + Cloud Run: 本番向けのサーバーレスデータ取込パイプライン構築

+----------------+       +-----------+       +---------------+       +---------------------+       +----------------+       +-------------------+
|                |       |           |       |               |       |                     |       |                |       |                   |
|  Event Source  +------>|  Pub/Sub  +------>|  Cloud Run    +------>|  BigQuery Storage   +------>|  BigQuery      +------>|  Downstream       |
| (Application,  |       |   Topic   |       |  (FastAPI     |       |    Write API        |       |  Table         |       |  Consumers        |
|  IoT, Webhook) |       |           |       |   Receiver)   |       |                     |       | (Partitioned,  |       | (Analytics, ML,   |
|                |       |           |       |               |       |                     |       |  Clustered)    |       |  Dashboards)      |
+----------------+       +-----------+       +---------------+       +---------------------+       +----------------+       +-------------------+
        ^                                            |                                                      ^
        |                                            |                                                      |
        |                                            +------------------------------------------------------+
        |                                                    (Error Handling, DLQ, Monitoring)
        |
        +-----------------------------------------------------------------------------------------------------
                                                              (Direct HTTP Ingestion - bypasses Pub/Sub)

{
  "writeStream": "projects/my-project/datasets/my_dataset/tables/my_table/streams/my_stream_id",
  "offset": {
    "value": "123"
  },
  "jsonRows": {
    "rows": [
      {"event_id": "abc-123", "event_timestamp": "2023-10-27T10:00:00Z", "payload": {"user_id": 123, "action": "click"}},
      {"event_id": "def-456", "event_timestamp": "2023-10-27T10:00:01Z", "payload": {"user_id": 456, "action": "view"}}
    ]
  }
}

import json
import time
from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types
from google.cloud.bigquery_storage_v1 import writer

def append_rows_to_bq_stream(
    project_id: str,
    dataset_id: str,
    table_id: str,
    rows: list[dict] # List of dictionaries, each representing a row
):
    """Appends rows to a BigQuery table using the Storage Write API."""

    client = bigquery_storage_v1.BigQueryWriteClient()
    table_path = client.table_path(project_id, dataset_id, table_id)

    # Create a write stream.
    # Use the 'PENDING' type for exactly-once semantics and explicit commit.
    write_stream = types.WriteStream(type_=types.WriteStream.Type.PENDING)
    write_stream = client.create_write_stream(
        parent=table_path, write_stream=write_stream
    )
    stream_name = write_stream.name
    print(f"Created write stream: {stream_name}")

    # Create a JSON stream writer.
    # This writer manages the connection and sends batches.
    # The `offset` is managed internally by the writer for exactly-once delivery.
    json_stream_writer = writer.JsonStreamWriter(
        client,
        stream_name,
        # The schema of the table. The Storage Write API uses this to validate
        # and correctly parse JSON rows.
        # It's crucial to provide the correct schema.
        table_schema={
            "fields": [
                {"name": "event_id", "type": "STRING", "mode": "REQUIRED"},
                {"name": "event_timestamp", "type": "TIMESTAMP", "mode": "REQUIRED"},
                {"name": "user_id", "type": "INTEGER", "mode": "NULLABLE"},
                {"name": "action", "type": "STRING", "mode": "NULLABLE"},
                {"name": "payload", "type": "JSON", "mode": "NULLABLE"}
            ]
        }
    )

    # Append rows. The writer handles batching and sending.
    # It also manages the offset for exactly-once semantics.
    try:
        # The `append_rows` method returns a future.
        # We wait for it to complete to ensure the write is acknowledged.
        # In a real application, you might manage these futures asynchronously.
        append_future = json_stream_writer.append_rows(rows)
        response = append_future.result() # Blocks until response is received

        if response.row_errors:
            print(f"Row errors encountered: {response.row_errors}")
        else:
            print(f"Successfully appended {len(rows)} rows to stream {stream_name} at offset {response.append_result.offset.value}")

    except Exception as e:
        print(f"Error appending rows: {e}")
        # In a real system, you'd handle this more robustly,
        # potentially retrying or sending to a DLQ.
        client.batch_finalize_write_stream(names=[stream_name]) # Finalize to release resources
        raise

    finally:
        # Finalize the stream. This makes the buffered data visible.
        # For PENDING streams, data is only visible after commit.
        # For COMMITTED streams (not used here), data is visible immediately.
        client.finalize_write_stream(name=stream_name)
        print(f"Finalized write stream: {stream_name}")

        # Commit the stream. This is required for PENDING streams
        # to make the data permanent.
        batch_commit_write_streams_request = types.BatchCommitWriteStreamsRequest()
        batch_commit_write_streams_request.parent = table_path
        batch_commit_write_streams_request.write_streams.append(stream_name)
        client.batch_commit_write_streams(batch_commit_write_streams_request)
        print(f"Committed write stream: {stream_name}")

# Example usage:
if __name__ == "__main__":
    PROJECT_ID = "your-gcp-project-id"
    DATASET_ID = "your_dataset"
    TABLE_ID = "your_table"

    sample_rows = [
        {"event_id": "uuid-1", "event_timestamp": "2023-10-27T10:00:00.123456Z", "user_id": 1, "action": "login", "payload": {"ip": "1.1.1.1"}},
        {"event_id": "uuid-2", "event_timestamp": "2023-10-27T10:00:01.678901Z", "user_id": 2, "action": "logout", "payload": {"duration_s": 600}},
    ]

    # Ensure the table and dataset exist before running this.
    # The table schema should match the fields provided to JsonStreamWriter.
    append_rows_to_bq_stream(PROJECT_ID, DATASET_ID, TABLE_ID, sample_rows)

CREATE TABLE `your-gcp-project-id.your_dataset.your_table`
(
    event_id STRING NOT NULL OPTIONS(description="Unique identifier for the event"),
    event_timestamp TIMESTAMP NOT NULL OPTIONS(description="Timestamp when the event occurred"),
    user_id INTEGER OPTIONS(description="Identifier for the user"),
    session_id STRING OPTIONS(description="Identifier for the user session"),
    action STRING OPTIONS(description="Type of action performed"),
    payload JSON OPTIONS(description="JSON payload containing additional event details")
)
PARTITION BY DATE(event_timestamp) -- Partition by the date component of event_timestamp
CLUSTER BY user_id, session_id      -- Cluster within each partition by user_id and session_id
OPTIONS(
    description="Table for storing application events, partitioned by event date and clustered by user/session.",
    partition_expiration_days=90, -- Partitions older than 90 days will be automatically deleted
    require_partition_filter=true -- Enforce partition filtering in queries to prevent full table scans
);

import os
import asyncio
import logging
from contextlib import asynccontextmanager
from typing import List, Dict, Any, Optional

from fastapi import FastAPI, Request, HTTPException, status
from pydantic import BaseModel, Field
from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types
from google.cloud.bigquery_storage_v1 import writer
from concurrent.futures import ThreadPoolExecutor

# --- Configuration ---
PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
DATASET_ID = os.environ.get("BQ_DATASET_ID")
TABLE_ID = os.environ.get("BQ_TABLE_ID")
BIGQUERY_TABLE_SCHEMA = {
    "fields": [
        {"name": "event_id", "type": "STRING", "mode": "REQUIRED"},
        {"name": "event_timestamp", "type": "TIMESTAMP", "mode": "REQUIRED"},
        {"name": "user_id", "type": "INTEGER", "mode": "NULLABLE"},
        {"name": "action", "type": "STRING", "mode": "NULLABLE"},
        {"name": "payload", "type": "JSON", "mode": "NULLABLE"}
    ]
}
MAX_BIGQUERY_STREAM_IDLE_SECONDS = 300 # Close streams if idle for too long
MAX_BIGQUERY_STREAM_BATCH_SIZE = 1000 # Max rows per append call
MAX_BIGQUERY_STREAM_BATCH_BYTES = 10 * 1024 * 1024 # 10MB per append call
MAX_WORKER_THREADS = 5 # For offloading synchronous BigQuery calls

# --- Logging ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Global BigQuery Client and Thread Pool ---
# We use a global client and thread pool to reuse resources efficiently.
bq_storage_client: Optional[bigquery_storage_v1.BigQueryWriteClient] = None
bq_executor: Optional[ThreadPoolExecutor] = None

# --- Pydantic Models for Request Validation ---
class EventPayload(BaseModel):
    ip: Optional[str] = None
    duration_s: Optional[int] = None
    # Add other dynamic fields as needed. Use Any for truly flexible JSON.
    # For strict validation, define all expected fields.

class Event(BaseModel):
    event_id: str = Field(..., description="Unique identifier for the event")
    event_timestamp: str = Field(..., description="ISO 8601 timestamp of the event")
    user_id: Optional[int] = Field(None, description="User ID associated with the event")
    action: Optional[str] = Field(None, description="Action performed by the user")
    payload: Optional[EventPayload] = Field(None, description="Additional event details")

# --- FastAPI Lifespan Management ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    global bq_storage_client, bq_executor
    logger.info("Starting up Cloud Run receiver...")

    if not all([PROJECT_ID, DATASET_ID, TABLE_ID]):
        logger.error("Missing required environment variables for BigQuery connection.")
        raise RuntimeError("BigQuery environment variables not set.")

    bq_storage_client = bigquery_storage_v1.BigQueryWriteClient()
    bq_executor = ThreadPoolExecutor(max_workers=MAX_WORKER_THREADS)
    logger.info("BigQueryWriteClient and ThreadPoolExecutor initialized.")

    yield # Application runs

    logger.info("Shutting down Cloud Run receiver...")
    if bq_storage_client:
        # In a real scenario, you'd manage open write streams here.
        # For this simple example, we assume streams are short-lived per request.
        # If using long-lived pooled streams, you'd finalize/commit them here.
        bq_storage_client.transport.close() # Close gRPC connections
        logger.info("BigQueryWriteClient closed.")
    if bq_executor:
        bq_executor.shutdown(wait=True)
        logger.info("ThreadPoolExecutor shut down.")
    logger.info("Cloud Run receiver shut down complete.")

app = FastAPI(
    title="BigQuery Serverless Ingestion Receiver",
    description="Receives events and streams them to BigQuery via Storage Write API.",
    version="1.0.0",
    lifespan=lifespan
)

# --- Idempotency Cache (In-memory for simplicity, use Redis for production) ---
# This is a very basic in-memory cache. For production, use a distributed cache like Redis.
# Stores {idempotency_key: (status, timestamp)}
IDEMPOTENCY_CACHE: Dict[str, tuple[str, float]] = {}
IDEMPOTENCY_CACHE_EXPIRATION_SECONDS = 3600 # 1 hour

def check_idempotency(idempotency_key: str) -> Optional[str]:
    """Checks idempotency cache. Returns 'processed' if seen, None otherwise."""
    if idempotency_key in IDEMPOTENCY_CACHE:
        status, timestamp = IDEMPOTENCY_CACHE[idempotency_key]
        if time.time() - timestamp < IDEMPOTENCY_CACHE_EXPIRATION_SECONDS:
            logger.info(f"Idempotent request detected for key: {idempotency_key}. Status: {status}")
            return status
        else:
            # Expired entry, remove it
            del IDEMPOTENCY_CACHE[idempotency_key]
    return None

def set_idempotency_status(idempotency_key: str, status: str):
    """Sets idempotency status."""
    IDEMPOTENCY_CACHE[idempotency_key] = (status, time.time())

# --- BigQuery Write Function (Synchronous, to be run in thread pool) ---
def _write_to_bigquery_sync(
    client: bigquery_storage_v1.BigQueryWriteClient,
    table_path: str,
    rows_to_insert: List[Dict[str, Any]]
):
    """Synchronously writes rows to BigQuery using Storage Write API."""
    write_stream = client.create_write_stream(
        parent=table_path, write_stream=types.WriteStream(type_=types.WriteStream.Type.PENDING)
    )
    stream_name = write_stream.name
    logger.info(f"Created write stream: {stream_name}")

    json_stream_writer = writer.JsonStreamWriter(
        client,
        stream_name,
        table_schema=BIGQUERY_TABLE_SCHEMA
    )

    try:
        append_future = json_stream_writer.append_rows(rows_to_insert)
        response = append_future.result() # Blocks until response

        if response.row_errors:
            logger.error(f"Row errors for stream {stream_name}: {response.row_errors}")
            raise RuntimeError(f"BigQuery row errors: {response.row_errors}")
        else:
            logger.info(f"Successfully appended {len(rows_to_insert)} rows to stream {stream_name} at offset {response.append_result.offset.value}")

    except Exception as e:
        logger.error(f"Failed to append rows to BigQuery stream {stream_name}: {e}")
        # Finalize to release resources even on error
        client.finalize_write_stream(name=stream_name)
        raise
    finally:
        # Finalize and commit the stream to make data visible
        client.finalize_write_stream(name=stream_name)
        commit_request = types.BatchCommitWriteStreamsRequest(
            parent=table_path,
            write_streams=[stream_name]
        )
        client.batch_commit_write_streams(commit_request)
        logger.info(f"Committed write stream: {stream_name}")


# --- FastAPI Endpoint ---
@app.post("/ingest")
async def ingest_events(request: Request, events: List[Event]):
    """
    Receives a batch of events and streams them to BigQuery.
    """
    idempotency_key = request.headers.get("Idempotency-Key")

    if idempotency_key:
        status_in_cache = check_idempotency(idempotency_key)
        if status_in_cache == "processed":
            return {"message": "Request already processed", "status": "idempotent_success"}
        elif status_in_cache == "processing":
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Request with this Idempotency-Key is already being processed."
            )
        set_idempotency_status(idempotency_key, "processing")

    if not bq_storage_client or not bq_executor:
        logger.error("BigQuery client or executor not initialized.")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Service not ready."
        )

    table_path = bq_storage_client.table_path(PROJECT_ID, DATASET_ID, TABLE_ID)
    rows_to_insert = [event.model_dump(mode='json') for event in events]

    try:
        # Offload synchronous BigQuery write to a separate thread
        await asyncio.to_thread(
            _write_to_bigquery_sync,
            bq_storage_client,
            table_path,
            rows_to_insert
        )
        if idempotency_key:
            set_idempotency_status(idempotency_key, "processed")
        return {"message": f"Successfully ingested {len(events)} events."}
    except Exception as e:
        logger.error(f"Failed to ingest events: {e}", exc_info=True)
        if idempotency_key:
            set_idempotency_status(idempotency_key, "failed")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to write to BigQuery: {e}"
        )

# Example of a health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "ok"}

# Use an official Python runtime as a parent image
FROM python:3.10-slim-buster

# Set the working directory in the container
WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy the application code
COPY . .

# Expose the port the app runs on
ENV PORT 8080

# Run the application with Uvicorn
CMD exec uvicorn main:app --host 0.0.0.0 --port $PORT --workers 1

fastapi
uvicorn[standard]
google-cloud-bigquery-storage
pydantic

gcloud run deploy bigquery-ingester \
  --image gcr.io/your-project-id/bigquery-ingester:latest \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated \
  --cpu 1 \
  --memory 2Gi \
  --concurrency 80 \
  --timeout 120s \
  --min-instances 1 \
  --max-instances 10 \
  --set-env-vars GCP_PROJECT_ID=your-project-id,BQ_DATASET_ID=your_dataset,BQ_TABLE_ID=your_table \
  --service-account bigquery-ingester-sa@your-project-id.iam.gserviceaccount.com

# main.tf

# -----------------------------------------------------------------------------
# Provider and Configuration
# -----------------------------------------------------------------------------
terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
  required_version = ">= 1.0"
}

# Variables for project and region
variable "project_id" {
  description = "The GCP project ID."
  type        = string
}

variable "region" {
  description = "The GCP region for resources."
  type        = string
  default     = "us-central1"
}

variable "dataset_id" {
  description = "BigQuery Dataset ID for ingestion."
  type        = string
  default     = "event_data"
}

variable "table_id" {
  description = "BigQuery Table ID for ingested events."
  type        = string
  default     = "app_events"
}

variable "cloud_run_service_name" {
  description = "Name of the Cloud Run service."
  type        = string
  default     = "bigquery-ingester"
}

variable "cloud_run_job_name" {
  description = "Name of the Cloud Run Job (v2 worker)."
  type        = string
  default     = "bq-maintenance-job"
}

variable "image_tag" {
  description = "Docker image tag for the Cloud Run service/job."
  type        = string
  default     = "latest" # In production, use specific version tags like v1.0.0
}

# -----------------------------------------------------------------------------
# Enable Required APIs
# -----------------------------------------------------------------------------
resource "google_project_service" "apis" {
  for_each = toset([
    "run.googleapis.com",
    "pubsub.googleapis.com",
    "bigquery.googleapis.com",
    "bigqueryconnection.googleapis.com", # Required for Storage Write API
    "bigquerystorage.googleapis.com",    # Required for Storage Write API
    "artifactregistry.googleapis.com",   # For Docker image storage
    "cloudbuild.googleapis.com",         # If using Cloud Build for CI/CD
    "iam.googleapis.com"                 # For managing service accounts
  ])
  project = var.project_id
  service = each.key
  disable_on_destroy = false
}

# -----------------------------------------------------------------------------
# Service Account and IAM
# -----------------------------------------------------------------------------
resource "google_service_account" "ingester_sa" {
  project      = var.project_id
  account_id   = "bigquery-ingester-sa"
  display_name = "Service Account for BigQuery Ingestion Cloud Run Service"
}

# Grant BigQuery Data Editor role for writing to the table
resource "google_project_iam_member" "bq_data_editor" {
  project = var.project_id
  role    = "roles/bigquery.dataEditor" # Allows writing data to tables
  member  = "serviceAccount:${google_service_account.ingester_sa.email}"
}

# Grant Pub/Sub Publisher role if Cloud Run pushes to Pub/Sub (e.g., for DLQ)
# Or if Cloud Run pulls from Pub/Sub via subscription
resource "google_project_iam_member" "pubsub_publisher" {
  project = var.project_id
  role    = "roles/pubsub.publisher"
  member  = "serviceAccount:${google_service_account.ingester_sa.email}"
  # Comment out if Cloud Run only receives HTTP and doesn't publish to Pub/Sub
}

# Allow Cloud Run to invoke the service (for unauthenticated access, this is 'allUsers')
# If you want authenticated access, use `google_cloud_run_service_iam_member`
# and specify which service accounts or users can invoke it.
resource "google_cloud_run_service_iam_member" "allow_unauthenticated" {
  project  = var.project_id
  location = var.region
  service  = google_cloud_run_service.ingester_service.name
  role     = "roles/run.invoker"
  member   = "allUsers"
  # Remove this block or change 'allUsers' to specific service accounts for authenticated access.
}

# -----------------------------------------------------------------------------
# Pub/Sub Topics (Main and Dead Letter)
# -----------------------------------------------------------------------------
resource "google_pubsub_topic" "main_events_topic" {
  project = var.project_id
  name    = "main-events-topic"
}

resource "google_pubsub_topic" "dlq_events_topic" {
  project = var.project_id
  name    = "dlq-events-topic"
}

# Pub/Sub Subscription (if Cloud Run pulls from Pub/Sub)
# This example assumes direct HTTP ingestion, but if using Pub/Sub as input,
# you'd create a push subscription to the Cloud Run service here.
/*
resource "google_pubsub_subscription" "cloud_run_push_subscription" {
  project = var.project_id
  name    = "cloud-run-ingester-sub"
  topic   = google_pubsub_topic.main_events_topic.name

  ack_deadline_seconds = 600 # Adjust based on processing time
  message_retention_duration = "604800s" # 7 days

  push_config {
    push_endpoint = google_cloud_run_service.ingester_service.status[0].url
    oidc_token {
      service_account_email = google_service_account.ingester_sa.email
    }
  }

  # Enable Dead Letter Queue
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.dlq_events_topic.id
    max_delivery_attempts = 5
  }

  # Enable exactly-once delivery (requires corresponding client-side support)
  enable_exactly_once_delivery = true
}
*/

# -----------------------------------------------------------------------------
# BigQuery Dataset and Table
# -----------------------------------------------------------------------------
resource "google_bigquery_dataset" "event_dataset" {
  project    = var.project_id
  dataset_id = var.dataset_id
  location   = var.region
  description = "Dataset for storing application event data."
}

resource "google_bigquery_table" "events_table" {
  project    = var.project_id
  dataset_id = google_bigquery_dataset.event_dataset.dataset_id
  table_id   = var.table_id
  deletion_protection = false # Set to true in production for critical tables

  schema = jsonencode([
    {
      name        = "event_id"
      type        = "STRING"
      mode        = "REQUIRED"
      description = "Unique identifier for the event"
    },
    {
      name        = "event_timestamp"
      type        = "TIMESTAMP"
      mode        = "REQUIRED"
      description = "Timestamp when the event occurred"
    },
    {
      name        = "user_id"
      type        = "INTEGER"
      mode        = "NULLABLE"
      description = "Identifier for the user"
    },
    {
      name        = "session_id"
      type        = "STRING"
      mode        = "NULLABLE"
      description = "Identifier for the user session"
    },
    {
      name        = "action"
      type        = "STRING"
      mode        = "NULLABLE"
      description = "Type of action performed"
    },
    {
      name        = "payload"
      type        = "JSON"
      mode        = "NULLABLE"
      description = "JSON payload containing additional event details"
    }
  ])

  time_partitioning {
    type                    = "DAY"
    field                   = "event_timestamp"
    expiration_ms           = 90 * 24 * 60 * 60 * 1000 # 90 days in milliseconds
    require_partition_filter = true
  }

  clustering = ["user_id", "session_id"] # Order matters for clustering

  labels = {
    env = "production"
    app = "ingestion"
  }
}

# -----------------------------------------------------------------------------
# Cloud Run Service (FastAPI Receiver)
# -----------------------------------------------------------------------------
resource "google_cloud_run_service" "ingester_service" {
  project  = var.project_id
  location = var.region
  name     = var.cloud_run_service_name

  template {
    spec {
      service_account_name = google_service_account.ingester_sa.email
      containers {
        image = "gcr.io/${var.project_id}/${var.cloud_run_service_name}:${var.image_tag}"
        env {
          name  = "GCP_PROJECT_ID"
          value = var.project_id
        }
        env {
          name  = "BQ_DATASET_ID"
          value = google_bigquery_dataset.event_dataset.dataset_id
        }
        env {
          name  = "BQ_TABLE_ID"
          value = google_bigquery_table.events_table.table_id
        }
        resources {
          cpu_idle = true # Optimize for CPU usage when idle
          limits = {
            cpu    = "1000m" # 1 CPU
            memory = "2Gi"   # 2 GB RAM
          }
        }
      }
      container_concurrency = 80 # Max concurrent requests per instance
    }

    metadata {
      annotations = {
        "autoscaling.knative.dev/minScale" = "1"
        "autoscaling.knative.dev/maxScale" = "10"
        "run.googleapis.com/client-name" = "terraform"
        "run.googleapis.com/vpc-access-connector" = "" # If using VPC Access
        "run.googleapis.com/vpc-access-egress" = "all" # If using VPC Access
      }
    }
  }

  traffic {
    percent         = 100
    latest_revision = true
  }

  timeouts {
    create = "10m"
    update = "10m"
  }

  lifecycle {
    ignore_changes = [
      template[0].metadata[0].annotations["autoscaling.knative.dev/minScale"],
      template[0].metadata[0].annotations["autoscaling.knative.dev/maxScale"],
    ]
  }
}

# -----------------------------------------------------------------------------
# Cloud Run Job (v2 Worker - e.g., for schema migrations or backfills)
# -----------------------------------------------------------------------------
resource "google_cloud_run_v2_job" "bq_maintenance_job" {
  project  = var.project_id
  location = var.region
  name     = var.cloud_run_job_name

  template {
    template {
      service_account_name = google_service_account.ingester_sa.email
      containers {
        image = "gcr.io/${var.project_id}/${var.cloud_run_job_name}:${var.image_tag}"
        env {
          name  = "GCP_PROJECT_ID"
          value = var.project_id
        }
        env {
          name  = "BQ_DATASET_ID"
          value = google_bigquery_dataset.event_dataset.dataset_id
        }
        env {
          name  = "BQ_TABLE_ID"
          value = google_bigquery_table.events_table.table_id
        }
        resources {
          limits = {
            cpu    = "1000m"
            memory = "512Mi"
          }
        }
      }
      timeout = "3600s" # 1 hour timeout for the job
      task_count = 1 # Number of parallel tasks
    }
    # This example does not define an execution template directly,
    # as it's typically triggered manually or by Cloud Scheduler.
  }
  labels = {
    env = "production"
    app = "bq-maintenance"
  }
}