数据追加至BigQuery或Google Sheet时向Pub/Sub发送新数据消息的可行性及实现方法咨询

Absolutely, your end-to-end workflow is totally feasible on Google Cloud—let’s walk through exactly how to build it, starting with solving your core pain point: triggering Pub/Sub messages when new rows land in Google Sheets or BigQuery.

This entire pipeline leverages Google Cloud’s native service integrations, so you don’t need any third-party tools to make it work. Every step (trigger → Pub/Sub → image processing → table update) can be built with GCP’s managed services, ensuring low maintenance and immediate execution as required.

For Google Sheets #

Sheets doesn’t have a native Pub/Sub trigger, but we can use Google Apps Script with an installable trigger to detect new rows and publish messages. Here’s how:

1. Open your Google Sheet, go to Extensions > Apps Script to launch the script editor.
2. Replace the default code with this snippet (customize the topic name and row logic to match your sheet):

function publishNewRowToPubSub() {
  const sheet = SpreadsheetApp.getActiveSheet();
  const lastRow = sheet.getLastRow();
  // Fetch the most recent row's data (adjust columns as needed)
  const newRow = sheet.getRange(lastRow, 1, 1, sheet.getLastColumn()).getValues()[0];
  
  // Package row data into a structured JSON payload
  const payload = JSON.stringify({
    source: "google_sheets",
    sheet_id: sheet.getId(),
    row_index: lastRow,
    row_data: newRow
  });
  
  // Publish to your Pub/Sub topic
  const topicFullName = "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME";
  PubSub.publish(topicFullName, Utilities.newBlob(payload).getBytes());
}

3. Set up an installable onChange trigger:
   • In the Apps Script editor, go to Edit > Current project's triggers.
   • Click Add trigger, set:
     • Choose which function to run: publishNewRowToPubSub
     • Choose which deployment to run: Head
     • Select event source: From spreadsheet
     • Select event type: On change
4. Grant permissions: The script will need OAuth access to your sheet and Pub/Sub. Follow the prompts to authorize it, then in your GCP console, add the Pub/Sub Publisher role to your Apps Script service account (format: YOUR_PROJECT_ID@gserviceaccount.com).

For Google BigQuery #

BigQuery has native event triggers via Eventarc (integrated with Cloud Functions) to detect row inserts. This works for both streaming inserts and batch loads:

1. Go to Cloud Functions in your GCP console and create a new function.
2. Configure the trigger:
   • Trigger type: Eventarc
   • Event provider: BigQuery
   • Event type: Table data inserted
   • Select your target dataset and table.
3. Use this Node.js code as your function (adjust the Pub/Sub topic and query logic):

const { PubSub } = require('@google-cloud/pubsub');
const { BigQuery } = require('@google-cloud/bigquery');
const pubsub = new PubSub();
const bigquery = new BigQuery();

exports.triggerPubSubOnInsert = async (event) => {
  const payload = JSON.parse(Buffer.from(event.data, 'base64').toString());
  const { table, timestamp } = payload;
  const tableFullId = `${table.projectId}.${table.datasetId}.${table.tableId}`;
  
  // Query the newly inserted rows (use _PARTITIONTIME for partitioned tables for efficiency)
  const query = `SELECT * FROM \`${tableFullId}\` WHERE _PARTITIONTIME >= TIMESTAMP("${timestamp}")`;
  const [rows] = await bigquery.query(query);
  
  // Publish each row to Pub/Sub (or batch them for efficiency)
  const topic = pubsub.topic('YOUR_TOPIC_NAME');
  for (const row of rows) {
    const message = JSON.stringify({
      source: "bigquery",
      table_id: tableFullId,
      row_data: row
    });
    await topic.publishMessage({ data: Buffer.from(message) });
  }
};

4. Assign permissions: Ensure your Cloud Function’s service account has the BigQuery Data Viewer and Pub/Sub Publisher roles.

Once messages are in Pub/Sub, create another Cloud Function to subscribe to the topic and handle the rest:

1. Create a Cloud Function with a Pub/Sub trigger pointing to your topic.
2. Use this code snippet to handle image download, GCS upload, and table updates (customize based on your sheet/BigQuery schema):

const { Storage } = require('@google-cloud/storage');
const { google } = require('googleapis');
const sheets = google.sheets('v4');
const { BigQuery } = require('@google-cloud/bigquery');
const storage = new Storage();
const bigquery = new BigQuery();

exports.processImageAndUpdateTable = async (message) => {
  const data = JSON.parse(Buffer.from(message.data, 'base64').toString());
  const { source, row_data, sheet_id, row_index, table_id } = data;
  
  // Extract the image URL from your row data (adjust index to match your column)
  const imageUrl = row_data[2];
  
  // Download the image
  const fetch = require('node-fetch');
  const response = await fetch(imageUrl);
  const imageBuffer = await response.buffer();
  
  // Upload to GCS
  const bucketName = "YOUR_GCS_BUCKET_NAME";
  const fileName = `processed_images/${Date.now()}-${Math.random().toString(36).slice(2)}.jpg`;
  const file = storage.bucket(bucketName).file(fileName);
  await file.save(imageBuffer);
  
  // Get the GCS path (or generate a signed URL for public access)
  const gcsPath = `gs://${bucketName}/${fileName}`;
  
  // Update the original table
  if (source === "google_sheets") {
    // Update the corresponding row in Sheets (adjust column range as needed)
    await sheets.spreadsheets.values.update({
      spreadsheetId: sheet_id,
      range: `Sheet1!D${row_index}`, // Column D stores the GCS link
      valueInputOption: 'RAW',
      requestBody: { values: [[gcsPath]] },
      auth: new google.auth.GoogleAuth({ scopes: ['https://www.googleapis.com/auth/spreadsheets'] })
    });
  } else if (source === "bigquery") {
    // Update BigQuery (use your table's primary key to target the row)
    const query = `UPDATE \`${table_id}\` SET gcs_image_link = @gcsPath WHERE id = @rowId`;
    await bigquery.query({
      query,
      params: { gcsPath, rowId: row_data.id }
    });
  }
  
  // Acknowledge the message to avoid reprocessing
  message.ack();
};

3. Assign permissions: Give this function’s service account the GCS Object Creator, Google Sheets Editor (if handling Sheets), and BigQuery Data Editor (if handling BigQuery) roles.

• Keep all resources in the same GCP project to simplify permission management.
• For large batch inserts in Sheets, be aware of Apps Script’s 6-minute execution limit—split large jobs if needed.
• In BigQuery, use partitioned tables and _PARTITIONTIME to efficiently fetch new rows without scanning the entire table.
• Add error handling (e.g., retry failed image downloads, dead-letter queues for Pub/Sub messages) to make the pipeline robust.

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