从Pubsub用BigQuery编写查询

问题描述

需要一些帮助.

我正在接收带有PubSub主题中的数据的消息，我需要插入从消息中获取的数据，并使用背景云功能(PUB/SUB)在BigQuery中进行查询...

I'm receiving messages with data in a PubSub topic, I need to insert the data I get from the message and query in BigQuery with a background cloud function(PUB/SUB)...

我要做什么:

/**
 * Triggered from a message on a Cloud Pub/Sub topic.
 *
 * @param {!Object} event Event payload.
 * @param {!Object} context Metadata for the event.
 */
exports.insertBigQuery = (message, context) => {
    extractQuery(message.data);

};

function extractQuery(pubSubMessage){
    // Decide base64 the PubSub message
    let logData = Buffer.from(pubSubMessage, 'base64').toString();
    // Convert it in JSON
    let logMessage= JSON.parse(logData);

    console.log(logMessage.customerToken)
    console.log(logMessage.fbclid)
    console.log(logMessage.fbc)
    console.log(logMessage.fbp)
    console.log(logMessage.firstHitTS)
    console.log(logMessage.consentFB)

    main();

    return logMessage

    }

"use strict";

function main() {
  const { BigQuery } = require("@google-cloud/bigquery");
  const bigquery = new BigQuery();

  async function query() {
    const query = `INSERT INTO MYTABLE( customerToken, fbclid, fbc, fbp, firstHitTS, consentFB)
    VALUES ("customerTokenSCRIPTCLOUD","fbclidSCRIPT"," fbcSCRIPTCLOUD"," fbpSCRIPTCLOUD","2021-01-05",TRUE )`;

    const options = {
      query: query,
      location: "US",
    };

    const [job] = await bigquery.createQueryJob(options);
    console.log(`Job ${job.id} started.`);

    const [rows] = await job.getQueryResults();

    console.log("Rows:");
    rows.forEach((row) => console.log(row));
  }

  query();
}

现在，每次我收到一条消息时，我都会在bigQuery中进行查询，但是我的VALUES是硬编码的，如您在此处看到的那样:

Now every time I receive a message I query in bigQuery, but my VALUES is hard coded, as you can see here:

const query = `INSERT INTO devsensetestprojects.TestDataSet.fbSimpleData( customerToken, fbclid, fbc, fbp, firstHitTS, consentFB)
    VALUES ("customerTokenSCRIPTCLOUD","fbclidSCRIPT"," fbcSCRIPTCLOUD"," fbpSCRIPTCLOUD","2021-01-05",TRUE )`;

我不能做的是从 function extractQuery(pubSubMessage)函数中获取值，并在查询中使用它们，就像在函数(logMessage.SOMEVALUE)中使用的方法一样我需要的正确值.

What I'm not able to do is to get the values from function extractQuery(pubSubMessage) and use them in my query the same way I use in the function (logMessage.SOMEVALUE) to have the correct values I need.

提前谢谢！

推荐答案

正如您所说，您是开发的初学者.这里有一个更简洁高效的代码.我没有测试过它，但是它离您想要的东西更近了.让我知道对您来说有些神秘！

As you said, you are beginner in development. Here a more concise and efficient code. I didn't tested it but it is closer of what you want. Let me know is some part are mysterious for you!

// Make them global to load them only when the Cloud Function instance is created
// They will be reused in the subsequent processing and until the instance deletion
const { BigQuery } = require("@google-cloud/bigquery");
const bigquery = new BigQuery();



exports.insertBigQuery = async (message, context) => {

    // Decode base64 the PubSub message
    let logData = Buffer.from(message.data, 'base64').toString();
    // Convert it in JSON
    let logMessage= JSON.parse(logData);

    const query = createQuery(logMessage)

    const options = {
        query: query,
        location: "US",
    };

    const [job] = await bigquery.createQueryJob(options);
    console.log(`Job ${job.id} started.`);

    // Only wait the end of the job. Theere is no row as answer, it's only an insert
    await job.getQueryResults();

}

function createQuery(logMessage) {
    // You maybe have to format correctly the logMessage.firstHitTS to be accepted by BigQuery as a date.
    return `INSERT INTO MYTABLE(customerToken, fbclid, fbc, fbp, firstHitTS, consentFB)
                   VALUES (logMessage.customerToken, logMessage.fbclid, logMessage.fbc, logMessage.fbp,
                           logMessage.firstHitTS, logMessage.consentFB)`;
}

这篇关于从Pubsub用BigQuery编写查询的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！