How to Get Started with JavaScript and InfluxDB

npm install @influxdata/influxdb-client
npm install @influxdata/influxdb-client-apis

<script type="module">
    // import latest release from npm repository 
    import {InfluxDB} from 'https://unpkg.com/@influxdata/influxdb-client/dist/index.browser.mjs'
<script>

• asciichart: Console ASCII line charts in pure JavaScript
• blessed: A curses-like library with a high-level terminal interface API ("curses" is a terminal control library for Unix-like systems)
• chalk: Terminal string styling done right
• eventsource: A pure JavaScript implementation of an EventSource client

None of these are required to use the InfluxDB JavaScript client.

Making a connection

Once dependencies are installed and environment variables are set, you can make a connection to the bucket. For this example, the writer.js and reader.js instantiates an InfluxDB object with the url and token read from the environment: new InfluxDB({url, token}) This object provides methods to instantiate the different API clients, such as Writer or Query.

Inserting data

To insert data in InfluxDB, you have to follow the line protocol. This defines a structure for the four elements that constitute a data point:

• Measurement: This is the name of the table in which you are going to insert the data
• Tag set: A comma-separated set of key = value of data properties
• Field set: A comma-separated set of key = value of data dimensions, the data can be float (default), integer, uinterger, string, Boolean
• Time stamp (optional): Unix time stamp

Luckily, the InfluxDB JavaScript client provides a wrapper class to be used with the Writer API to insert one or many data points. The following code shows how to connect to the event source and for each message instantiate a point to be added to the bucket:

#!/usr/bin/env node

const {InfluxDB, Point, HttpError} = require('@influxdata/influxdb-client');
const {url, token, org, bucket, sseUrl} = require('./env');
var EventSource = require('eventsource');
const {hostname} = require('os');

//Creates a writer with "seconds" as precision
const writeApi = new InfluxDB({url, token}).getWriteApi(org, bucket, 's');
//Sets the common tags for this writer
writeApi.useDefaultTags({location: hostname(), source:'wikimedia', sseUrl:sseUrl });

console.log(`Connecting to EventStreams at ${sseUrl}`);
var eventSource = new EventSource(sseUrl);

eventSource.onmessage = function(event) {
    // event.data will be a JSON string containing the message event.
    const d = JSON.parse(event.data);
    if( d.length != undefined ){
        const dataPoint = new Point('edition')
            .tag( 'user',d.user )
            .tag( 'isBot', d.bot )
            .floatField('value', d.length.new)
        writeApi.writePoint( dataPoint );
        console.log( dataPoint );
        writeApi
        .flush().then(() => {})
        .catch(e => {
            console.log('\nFinished ERROR: ' + e);
        });
    }
};

Notice that in lines 9–11 the writeApi is instantiated with the organization and bucket environment variables, but also with a precision value. In this case it's seconds, which defines the granularity of the time stamp dimension of the data to be written; check the other possible values here. Also, you can set some default tags for each data point to be written — the source of the data in this case. Also notice that the client gives you the option to extend this default tag set with other values; for each data point the user and isBot tags are added. The field written is the length of the change made (line 23), the writePoint method accepts the Point instance, and then the writeApi instance is flushed. You should always close the writeApi in order to flush pending changes and close pending scheduled retry executions. The writer.js runs indefinitely for each message pushed by the event source.

Querying data

Once the data is in the bucket, the InfluxDB JavaScript client provides another API client to query the data. In this example, the InfluxDB object runs a query that returns the number of data points grouped by the isBot tag in the last 10 seconds:

const query = '\
from(bucket:"js-sample")\
|> range(start: -10s)\
|> filter(fn:(r) => r._measurement == "edition")\
|> group(columns: ["isBot"])\
|> count()\
';

The query is written in the Flux functional data scripting language. This is designed for querying, analyzing and acting on data over InfluxDB 2.0. The previous language, the InfluxDB SQL-like query language, is still supported at the /query compatibility endpoint of the API, but the recommendation is to use the full power of the new language. As you can see, the query defines the bucket, range of data, filters to be applied, grouping columns and data functions to be applied. The following function shows you how to run the query against the QueryApi:

function queryExample(fluxQuery) {
    const queryApi = new InfluxDB({url, token}).getQueryApi(org)
    queryApi.queryRows(fluxQuery, {
        next(row, tableMeta) {
            const o = tableMeta.toObject(row);
            pushRow(o);
            render();
        },
        complete() { 
            console.log('FINISHED')
        },
        error(error) {
            console.log('QUERY FAILED', error)
        },
    });
}

For each row returned, the data is parsed into an object and passed to a function that stores it in an array based on the isBot attribute. The two arrays of the series, bots and humans, are then rendered using the asciichart library. The full code for the reader is shown below:

const {InfluxDB} = require('@influxdata/influxdb-client');
const {url, token, org, bucket, chart, screen} = require('./env');
const chalk = require('chalk');
var asciichart = require ('asciichart');
const maxLength = 100;
const bots = [];
const humans = [];
var config = {
    height:  18,         // any height you want
    colors: [
        asciichart.blue,
        asciichart.red,
        asciichart.default, // default color
        undefined, // equivalent to default
    ]
}

const query = `\
from(bucket:"${bucket}")\
|> range(start: -10s)\
|> filter(fn:(r) => r._measurement == "edition")\
|> group(columns: ["isBot"])\
|> count()\
`;

demo();

async function demo() {
    for (let index = 0; index < 300; index++) {
        queryExample( query );
        await sleep(500);
    }
}

function queryExample(fluxQuery) {
    const queryApi = new InfluxDB({url, token}).getQueryApi(org)
    queryApi.queryRows(fluxQuery, {
        next(row, tableMeta) {
            const o = tableMeta.toObject(row);
            pushRow(o);
            render();
        }, complete() {}
        , error(error) {
            console.log('QUERY FAILED', error)
        }
    });
}

function pushRow(row) {
    if (bots.length >= maxLength) {
        bots.shift ();
    }
    if (humans.length >= maxLength) {
        humans.shift ();
    }

    row.isBot == 'true' ? bots.push( row['_value']) : humans.push( row['_value']);
}

function render(){
    if(bots.length != 0 && humans.length != 0) {
        const plt = asciichart.plot ([bots,humans], config).split ('\n');
        chart.setLine (0, chalk.blue('Bots: '+bots[bots.length-1]) + ' ' + chalk.red('Humans: '+humans[humans.length-1]));
        plt.forEach ((line, i) => {
            chart.setLine (i + 1, line);
        });
    }
    screen.render();
}

function sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
}

Additional documentation and functionality

The InfluxDB JavaScript client also provides additional functionality (like the HealthAPI, among other wrapper classes) that simplifies building a production-ready pipeline. Check the client and InfluxDB API documentation for more details.

Conclusion

Time series databases like InfluxDB provide specialized functionality and tools to store and analyze data points with a high rate of ingestion, such as Internet of Things (IoT) data. This example functions as a useful lens to understand the functionality of the InfluxDB JavaScript client to collect high-volume streaming data. The client is a simple, standard Node.js module that lets you write and read data into or from an InfluxDB instance frictionlessly.

About the author:

Nicolas Bohorquez is a data architect at Merqueo, has been part of development teams in a handful of startups, and has founded three companies in the Americas. He is passionate about the modeling of complexity and the use of data science to improve the world.