HadoopのJob Trackerのログをtsvファイルに変換するrubyスクリプト

背景

hadoopでErrorが発生したが、原因が簡単に特定できない。mapフェイズでコケているのに、job trackerのログの最後には、「# of failed Reduce Tasks exceeded allowed limit\. FailedCount: 1\. LastFailedTask: task_*****_0001_r_000030」というエラーが表示される（タスクの名前にrとあるのがreduceタスクである証拠）。 Hadoop MapReduce Job の History ログの見方のスクリプトは素晴らしい。ログファイルを簡単にしてくれる。ただし今回の場合ホスト名がないと困ったので自分で作ることにした。

そのスクリプト

#!/usr/bin/ruby -Ku

task_hash = Hash.new{|h,tid|h[tid]={}}
attempt_hash = Hash.new{|h,tid|h[tid]={}}

def format_value(output_hash, input_hash, tag_list)
  if tid = input_hash["TASKID"]
    return if input_hash["TASK_ATTEMPT_ID"] and input_hash["TASK_ATTEMPT_ID"] !~ /_0$/
    tag_list.each do |tag|
      if input_hash[tag]
        v = input_hash[tag]
        v =  $1 if tag == "HOSTNAME" and v =~ /(\d+.*\.\d+.*\.\d+.*\.\d+)/
        output_hash[tid][tag] = v
      end
    end
  end
end

while line = gets
  line.chomp!
  hash = {}
  array = line.split(" ")
  log_type = array.shift
  array.each do |e|
    hash[$1]=$2 if e =~ /(.+)="(.+)"$/
  end
  case log_type
  when 'Task'
    format_value(task_hash, hash, %w(TASK_TYPE START_TIME TASK_STATUS FINISH_TIME))
  when 'MapAttempt'
    format_value(attempt_hash, hash, %w(START_TIME TASK_STATUS FINISH_TIME HOSTNAME TASK_ATTEMPT_ID))
  when 'ReduceAttempt'
    format_value(attempt_hash, hash, %w(START_TIME TASK_STATUS FINISH_TIME SHUFFLE_FINISHED SORT_FINISHED HOSTNAME TASK_ATTEMPT_ID))
  end
end

task_hash.each do |tid, task|
  task.merge!(attempt_hash[tid])
  a = [tid]
  %w(TASK_TYPE START_TIME FINISH_TIME TASK_STATUS HOSTNAME).each do |tag|
    a.push(task[tag])
  end
  puts a.join("\t")
end