集成 BigQuery SubPub 和 Cloud Functions

Question

我在一个项目中，我们需要使用 BigQuery、PubSub、Logs explorer 和 Cloud Functions。

项目：

每次发生特定事件（例如用户接受 cookie）时，系统都会向 BigQuery 中插入一个新查询，其中包含许多列（参数），例如：utm_source、utm_medium、consent_cookies 等...

在我的表中有这个新查询后，我需要读取列并获取要在云函数中使用的值。

在云函数中，我想使用这些值进行 api 调用。

到目前为止我能做到的事情：

我创建了一个日志路由接收器，用于过滤新条目并将日志发送到我的 PubSub 主题。

我被困在哪里：

我想创建一个 Cloud 函数，该函数在每次有新日志进入时触发，并且在该函数中我想访问日志中包含的信息，例如 utm_source、utm_medium、consent_cookies 等...并使用进行 api 调用的值。

有人可以帮助我吗？非常感谢！

我做了一个项目来说明流程：

1. 插入表格：

2.从此插入创建日志接收器：（过滤）

1. 现在，每次我创建一个新查询时，它都会转到 PUB/SUB，我会获取该查询的日志

4. 我想做的是触发一个关于这个主题的函数，并使用我在查询中的值来执行调用 api 等操作......

到目前为止，我能够编写此代码：

"use strict";

function main() {
  // Import the Google Cloud client library
  const { BigQuery } = require("@google-cloud/bigquery");

  async function queryDb() {
    
    const bigqueryClient = new BigQuery();

   
    const sqlQuery = `SELECT *  FROM \`mydatatable\``;

    const options = {
      query: sqlQuery,
      location: "europe-west3",
    };

    // Run the query
    const [rows] = await bigqueryClient.query(options);

    rows.forEach((row) => {
      const username = row.user_name;
    });
  }

  queryDb();
}

main();

现在我又被卡住了，我不知道如何从我创建的接收器中获取正确的查询并使用这些信息来拨打我的电话...

Answer 1

您有 2 个解决方案可以从 PubSub 消息中调用您的 Cloud Functions

• HTTP 功能：您可以设置 HTTP 调用。在 trigger-http 中创建您的 Cloud Function，并在您的 PubSub 主题上创建一个推送订阅以调用 Cloud Functions。不要忘记添加安全性（将您的函数设为私有并在 PubSub 上启用安全性），因为您的函数可以公开访问
• 后台函数：您可以将 Cloud Functions 直接绑定到 PubSub 主题。订阅会自动创建并链接到 Cloud Functions。安全性是内置的。

而且，因为您有 2 种类型的函数，所以您有 2 个不同的函数签名。我提供给你，处理是（相当）相同的。

function extractQuery(pubSubMessage){
    // Decide base64 the PubSub message
    let logData = Buffer.from(pubSubMessage, 'base64').toString();
    // Convert it in JSON
    let logMessage= JSON.parse(logData)
    // Extract the query from the log entry
    let query = logMessage.protoPayload.serviceData.jobInsertRequest.resource.jobConfiguration.query.query

    console.log(query)
    return query
}

// For HTTP functions
exports.bigqueryQueryInLog = (req, res) => {

    console.log(req.body)
    const query = extractQuery(req.body.message.data)

    res.status(200).send(query);
}

// For Background functions
exports.bigqueryQueryInLogTopic = (message, context) => {
    extractQuery(message.data)
};

记录的query 是您在日志条目中的insert into...。然后，您必须解析您的 SQL 请求以提取您想要的部分。