Kafka to InfluxDB ingestion

ingest.js

const Influx = require('influx');
const kafka = require('kafka-node'),
    ConsumerGroup = kafka.ConsumerGroup;

const consumer = new ConsumerGroup({
  kafkaHost: `${process.env.KAFKA_HOST||"localhost"}:9092`,
  groupId: 'price-history-trade-consumer'
}, ['trade-stream']);


const influx = new Influx.InfluxDB({
 host: process.env.INFLUX_HOST || 'localhost',
 database: 'price_history_db',
 schema: [
   {
     measurement: 'trade',
     fields: {
       price: Influx.FieldType.FLOAT,
       size: Influx.FieldType.INTEGER
     },
     tags: [
       'symbol',
       'side'
     ]
   }
 ]
});

consumer.on('message', function (message) {
  const trade = JSON.parse(message.value.toString('utf8'));
  console.log('DEBUG: Got new trade', trade);
  influx.writePoints([
    {
      measurement: 'trade',
      tags: {
        symbol: trade.symbol,
        side: trade.side
      },
      fields: { 
        price: trade.price, size: trade.size
      },
      timestamp: trade.timestamp || (new Date()).getTime() * 1000
    }
  ], {
    precision: 'u'
  });
});