Hadoop Streaming with Node.js

map.js

#!/usr/bin/env node

var stdin = process.openStdin();
var stdout = process.stdout;
var input = '';

stdin.setEncoding('utf8');
stdin.on('data', function(data) {
  if (data) {
    input += data;
    while (input.match(/\r?\n/)) {
      input = RegExp.rightContext;
      proc(RegExp.leftContext);
    }
  }
});

stdin.on('end', function() {
  if (input) {
    proc(input);
  }
});

function proc(line) {
  var words = line.split(' ');
  stdout.write(words[8] + ',1\n');
}

map.rb

#!/usr/bin/env ruby

ARGF.each do |line|
  line.chomp!
  words = line.split(' ')
  printf("%s,1\n", words[8])
end

reduce.js

#!/usr/bin/env node

var stdin = process.openStdin();
var stdout = process.stdout;
var counter = {};
var input = '';

stdin.setEncoding('utf8');
stdin.setEncoding('utf8');
stdin.on('data', function(data) {
  if (data) {
    input += data;
    while (input.match(/\r?\n/)) {
      input = RegExp.rightContext;
      proc(RegExp.leftContext);
    }
  }
});

stdin.on('end', function() {
  if (input) proc(input);
  for (var k in counter) {
    stdout.write(k + ':' + counter[k] + '\n');
  }
});

function proc(line) {
  var words = line.split(',');
  var word = words[0];
  var count = parseInt(words[1]);
  if (!counter[word]) counter[word] = 1;
  else counter[word] += count;
}

reduce.rb

#!/usr/bin/env ruby

counter = Hash.new
ARGF.each do |line|
  line.chomp!
  words = line.split(',')
  word = words[0]
  count = words[1].to_i
  if !counter.key?(word)
    counter[word] = 1
  else
    counter[word]+=count
  end
end

counter.each do |k, v|
  printf("%s:%d\n", k, v)
end