GCP Integration & Streaming Fundamentals¶

# Enable message ordering
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

# Publish with ordering key
future = publisher.publish(
    topic_path,
    data.encode("utf-8"),
    ordering_key="user123"  # Messages with same key are ordered
)

# Publish with attributes
future = publisher.publish(
    topic_path,
    data.encode("utf-8"),
    event_type="user_signup",
    region="us-central1",
    user_tier="premium"
)

# Subscription filter
subscription_filter = 'attributes.event_type="user_signup" AND attributes.user_tier="premium"'

# Create subscription with dead letter policy
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)
dead_letter_topic = publisher.topic_path(project_id, dead_letter_topic_id)

subscription = subscriber.create_subscription(
    request={
        "name": subscription_path,
        "topic": topic_path,
        "dead_letter_policy": {
            "dead_letter_topic": dead_letter_topic,
            "max_delivery_attempts": 5
        }
    }
)

# Enable exactly-once delivery
subscription = subscriber.create_subscription(
    request={
        "name": subscription_path,
        "topic": topic_path,
        "enable_exactly_once_delivery": True
    }
)

import functions_framework
from google.cloud import storage

@functions_framework.cloud_event
def process_file(cloud_event):
    """Process file upload to Cloud Storage"""

    # Extract event data
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]

    # Process the file
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    # Download and process file content
    content = blob.download_as_text()
    processed_content = content.upper()  # Simple processing

    # Upload processed file
    output_blob = bucket.blob(f"processed_{file_name}")
    output_blob.upload_from_string(processed_content)

    print(f"Processed {file_name}")

@functions_framework.cloud_event
def reliable_function(cloud_event):
    try:
        # Function logic
        process_data(cloud_event.data)

    except RetryableError as e:
        # Log error and let Pub/Sub retry
        print(f"Retryable error: {e}")
        raise  # Function will be retried

    except FatalError as e:
        # Log error but don't retry
        print(f"Fatal error: {e}")
        return  # Acknowledge message to prevent retry

import functions_framework
from cloudevents.http import CloudEvent

@functions_framework.cloud_event
def handle_storage_event(cloud_event: CloudEvent):
    """Handle Cloud Storage events via Eventarc"""

    # Extract event data
    data = cloud_event.data
    bucket = data["bucket"]
    name = data["name"]
    event_type = cloud_event["type"]

    print(f"Processing {event_type} for {name} in {bucket}")

    # Process based on event type
    if event_type == "google.cloud.storage.object.v1.finalized":
        process_new_file(bucket, name)
    elif event_type == "google.cloud.storage.object.v1.deleted":
        cleanup_references(bucket, name)

from google.cloud import tasks_v2

# Create task queue
client = tasks_v2.CloudTasksClient()
parent = f"projects/{project_id}/locations/{location_id}"

queue = {
    "name": f"{parent}/queues/my-queue",
    "rate_limits": {
        "max_dispatches_per_second": 10,
        "max_burst_size": 100,
        "max_concurrent_dispatches": 50
    },
    "retry_config": {
        "max_attempts": 5,
        "max_retry_duration": {"seconds": 3600},
        "min_backoff": {"seconds": 1},
        "max_backoff": {"seconds": 600},
        "max_doublings": 16
    }
}

response = client.create_queue(parent=parent, queue=queue)

# Create HTTP task
task = {
    "http_request": {
        "http_method": tasks_v2.HttpMethod.POST,
        "url": "https://my-service.run.app/process",
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"user_id": "123", "action": "send_email"}).encode()
    },
    "schedule_time": timestamp_pb2.Timestamp(
        seconds=int(time.time()) + 300  # Schedule 5 minutes from now
    )
}

# Add task to queue
response = client.create_task(
    parent=queue_path,
    task=task
)

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

def run_pipeline():
    pipeline_options = PipelineOptions([
        '--project=my-project',
        '--region=us-central1',
        '--runner=DataflowRunner',
        '--temp_location=gs://my-bucket/temp',
        '--staging_location=gs://my-bucket/staging'
    ])

    with beam.Pipeline(options=pipeline_options) as pipeline:
        (pipeline
         | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(topic='projects/my-project/topics/input')
         | 'Parse JSON' >> beam.Map(json.loads)
         | 'Filter Events' >> beam.Filter(lambda x: x.get('event_type') == 'purchase')
         | 'Transform Data' >> beam.Map(transform_purchase_data)
         | 'Window Data' >> beam.WindowInto(beam.window.FixedWindows(60))  # 1-minute windows
         | 'Aggregate' >> beam.CombinePerKey(sum)
         | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
             table='my-project:dataset.purchase_summary',
             write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
         ))

def transform_purchase_data(element):
    return (element['product_category'], element['amount'])