Create a Data Pipeline for MongoDB Change Stream Using Pub/Sub BigQuery Subscription

1
{
2
 "type" : "record",
3
 "name" : "Avro",
4
 "fields" : [
5
   {
6
     "name" : "id",
7
     "type" : "string"
8
   },
9
   {
10
     "name" : "source_data",
11
     "type" : "string"
12
   },
13
   {
14
     "name" : "Timestamp",
15
     "type" : "string"
16
   }
17
 ]
18
}

1
bq add-iam-policy-binding \
2
    --member="serviceAccount:service<project-number>@gcp-sa-pubsub.iam.gserviceaccount.com" \
3
    --role=roles/bigquery.dataEditor \
4
    -t "<dataset>.<table>"

1
npm install mongodb @google-cloud/pubsub avro-js

1
{
2
 "type": "record",
3
 "name": "DocumentMessage",
4
 "fields": [
5
   {
6
     "name": "id",
7
     "type": "string"
8
   },
9
   {
10
     "name": "source_data",
11
     "type": "string"
12
   },
13
   {
14
     "name": "Timestamp",
15
     "type": "string"
16
   }
17
 ]
18
}

1
import { MongoClient } from 'mongodb';
2
import { PubSub } from '@google-cloud/pubsub';
3
import avro from 'avro-js';
4
import fs from 'fs';
5
 
6
const MONGODB_URI = '<mongodb-connection-string>';
7
const PUB_SUB_TOPIC = 'projects/<project-name>/topics/<topic-name>';

1
let mongodbClient;
2
try {
3
   mongodbClient = new MongoClient(MONGODB_URI);
4
   await monitorCollectionForInserts(mongodbClient, 'my-database', 'my-collection');
5
} finally {
6
   mongodbClient.close();
7
}
8
 
9
async function monitorCollectionForInserts(client, databaseName, collectionName, timeInMs) {
10
   const collection = client.db(databaseName).collection(collectionName);
11
   // An aggregation pipeline that matches on new documents in the collection.
12
   const pipeline = [ { $match: { operationType: 'insert' } } ];
13
   const changeStream = collection.watch(pipeline);
14
 
15
   changeStream.on('change', event => {
16
       const document = event.fullDocument;
17
       publishDocumentAsMessage(document, PUB_SUB_TOPIC);
18
   });
19
 
20
   await closeChangeStream(timeInMs, changeStream);
21
}
22
 
23
function closeChangeStream(timeInMs = 60000, changeStream) {
24
   return new Promise((resolve) => {
25
       setTimeout(() => {
26
           console.log('Closing the change stream');
27
           changeStream.close();
28
           resolve();
29
       }, timeInMs)
30
   })
31
};

1
async function publishDocumentAsMessage(document, topicName) {
2
   const pubSubClient = new PubSub();
3
   const topic = pubSubClient.topic(topicName);
4
 
5
   const definition = fs.readFileSync('./document-message.avsc').toString();
6
   const type = avro.parse(definition);
7
 
8
   const message = {
9
       id: document?._id?.toString(),
10
       source_data: JSON.stringify(document),
11
       Timestamp: new Date().toISOString(),
12
   };
13
 
14
   const dataBuffer = Buffer.from(type.toString(message));
15
   try {
16
       const messageId = await topic.publishMessage({ data: dataBuffer });
17
       console.log(`Avro record ${messageId} published.`);
18
   } catch(error) {
19
       console.error(error);
20
   }
21
}

1
node ./index.mjs