基于Python脚本通过Pub/Sub批量转存消息至GCS的代码修改咨询

Hey there! Let's walk through exactly how to modify your existing Pub/Sub Python script to batch messages into files and upload them to Google Cloud Storage. Here's a step-by-step breakdown tailored to your setup:

1. First, Locate the Message Handling Core of Your Script #

Your current script almost certainly has a section where it receives and processes individual Pub/Sub messages. This is usually either:

• A callback function passed to subscriber.subscribe(), or
• A loop where you pull messages directly with subscriber.pull()

This is the key area you'll modify—since this is where you have access to each incoming message.

For reference, a typical existing callback might look like this:

def callback(message):
    # Your current logic for handling a single message
    print(f"Received message: {message.data.decode('utf-8')}")
    message.ack()

2. Add Batching Logic to Collect Messages #

First, add some global (or class-level, if you're using OOP) variables to track your batch, plus thresholds for when to trigger an upload:

from collections import deque
import time

# Configure your batch rules (tweak these based on your needs)
BATCH_SIZE = 100  # Upload after 100 messages
BATCH_TIMEOUT = 30  # Or upload after 30 seconds, whichever comes first
message_batch = deque()
last_batch_upload_time = time.time()

Then, modify your existing callback to collect messages into the batch instead of processing them one-by-one, and check if it's time to upload:

def callback(message):
    global message_batch, last_batch_upload_time
    
    # Add the message content to your batch (decode bytes to string if needed)
    message_content = message.data.decode("utf-8")
    message_batch.append(message_content)
    
    # Check if we hit either the size or time threshold
    current_time = time.time()
    if len(message_batch) >= BATCH_SIZE or (current_time - last_batch_upload_time) >= BATCH_TIMEOUT:
        # Process the batch and upload to GCS
        upload_batch_to_gcs(message_batch)
        # Reset the batch and timer
        message_batch.clear()
        last_batch_upload_time = current_time
    
    # Acknowledge the message so it's not re-delivered
    message.ack()

3. Implement the Batch Upload to GCS Function #

Add this helper function to your script—this is where you'll convert the batch into a file and send it to Cloud Storage. Make sure you have the google-cloud-storage package installed (pip install google-cloud-storage):

from google.cloud import storage
import tempfile
import uuid
import os

def upload_batch_to_gcs(batch):
    # Create a unique filename to avoid overwrites
    batch_filename = f"pubsub_batch_{uuid.uuid4().hex}.txt"
    
    # Write the batch to a temporary local file
    with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_file:
        for msg in batch:
            temp_file.write(f"{msg}\n")  # Separate messages with newlines for readability
    
    # Upload the file to your GCS bucket
    storage_client = storage.Client()
    bucket = storage_client.bucket("your-gcs-bucket-name")  # Replace with your bucket name
    blob = bucket.blob(f"batched_messages/{batch_filename}")  # Path within your bucket
    blob.upload_from_filename(temp_file.name)
    
    # Clean up the temporary file
    os.unlink(temp_file.name)
    
    print(f"Successfully uploaded batch to: gs://your-gcs-bucket-name/batched_messages/{batch_filename}")

4. Handle Edge Cases (Don't Forget These!) #

• Remaining messages on script shutdown: Add logic to upload any leftover messages in the batch when you stop the script. For example:

  try:
      print("Listening for messages... Press Ctrl+C to stop.")
      while True:
          time.sleep(60)
  except KeyboardInterrupt:
      # Upload any remaining messages before exiting
      if message_batch:
          upload_batch_to_gcs(message_batch)
      subscriber.close()
      print("Stopped listener and processed remaining messages.")
  

• Error handling: Wrap the GCS upload in a try-except block to handle upload failures (e.g., retry the upload, or nack the messages so they get re-delivered later).

5. Exact Placement in Your Script #

If your existing script follows a standard structure, here's where each piece goes:

1. Batch configuration variables: Place these right after initializing your Pub/Sub subscriber.
2. Modified callback function: Replace your existing callback with the updated version that collects batches.
3. upload_batch_to_gcs function: Add this anywhere in the script (before the callback is ideal, so it's defined when the callback runs).
4. Shutdown logic: Insert the try-except block that handles remaining messages where your script's main loop runs.

内容的提问来源于stack exchange，提问作者Srinivas Rao