gistfile1.txt

var pg = require('pg');
var QueryStream = require('pg-query-stream');
var JSONStream = require('JSONStream');
queue = require('kue').createQueue();

pg.connect(function(err, client, done) {
  if (err) throw err;
  var query = new QueryStream(
    'select address from voter_addr where status is null;',
  );
  var stream = client.query(query);

  stream.on('end', done);
  var batch = [];
  stream.pipe(row => {
    batch.push(row.rows[0].address);
    if (set.length === 1000) {
      queue
        .create('batch', {
          batch,
        })
        .save();
      batch = [];
    }
  });
});

queue.process('batch', function(job, done) {
  geocode(batch, done);
});