hdfs.js

var through = require('through');
var path = require('path');
var ss = require('stream-stream');
var avro = require('node-avro-io');

var hdfs = require('webhdfs').createClient({
  host: '' // stuff goes here
});

avrostream2('/log/ap/2014-04-01/01/ClientEvent', hdfs)
  .pipe(new avro.DataFile.Reader())
  .on('data', console.log.bind(console))


function avrostream2(partpath, hdfsClient) {
  var out = ss();
  hdfs.readdir(partpath, function(err, files) {
    if (err) {
      return emitErr(err);
    }

    if (!files.length) {
      return emitErr(new Error('No files found at ' + partpath + '\n' + JSON.stringify(files)));
    }

    var fstreams = files.map(function(fstat, i) {
      var fname = fstat.pathSuffix;
      var fstream = hdfsClient.createReadStream(path.join(partpath, fname));
      fstream.pause();
      out.write(fstream);
      return fstream;
    });

    fstreams[0].resume();
  })

  function emitErr(err) {
    out.emit('error', err)
    out.end();
    return;
  }
  
  return out;
}