CogSpark CogSpark

PubSub to BQ function

Table of Contents

Cloud Function: PubSub to Big Query

Create the pubsub topic

gcloud pubsub topics create myTopic

Create the BQ Schema

schema.json

[
    {"name": "device_id", "type": "string", "mode": "nullable"},
    {"name": "time", "type": "string", "mode": "nullable"},
    {"name": "temp", "type": "string", "mode": "nullable"}
  ]

bq mk mydataset bq mk -t mydataset.temperature ./schema.json

Deploy the function

package.json

{
    "name": "sample-pubsub",
    "version": "0.0.1",
    "dependencies": {
      "@google-cloud/bigquery": "*"
   
    }
  }

index.js

/**
 * Triggered from a message on a Cloud Pub/Sub topic.
 *
 * @param {!Object} event The Cloud Functions event.
 * @param {!Function} The callback function.
 */
const BigQuery = require('@google-cloud/bigquery');

exports.helloPubSub = (event, callback) => {
  const pubsubMessage = event.data;
  result = Buffer.from(pubsubMessage.data, 'base64').toString()
  jresult=JSON.parse(result);
  const projectId = 'crg-gcp';
  
  // Creates a client
  const bigquery = new BigQuery({
    projectId: projectId,
  });
  const dataset = bigquery.dataset("mydataset");
  const table = dataset.table("temperature");


console.log(table.insert({
    device_id: jresult.device_id,
    temp: jresult.temperature,
    time: jresult.time
  })
);

  callback();
};

Deploy the function

gsutil mb gs://crg-functions-stage-bucket
gcloud beta functions deploy helloPubSub \
    --stage-bucket crg-functions-stage-bucket \
    --trigger-topic myTopic

Test the system

gcloud pubsub topics publish myTopic --message "{'device_id':'1','temperature':'2','time':'3'}"

Cleanup

gcloud pubsub topics delete myTopic