translate.rb

#!/usr/bin/env ruby

require 'rubygems'
require 'json'
require File.dirname(__FILE__)+'/app/helpers/graph.rb'


class P2jTranslator
  attr_accessor :nodeGroups, :json
  
  def initialize plan_package
    @nodeGroups = Hash.new{|h,k| h[k] = {}}
    result = {
      :id   => plan_package[:uuid],
      :name => plan_package[:jobName],
      :properties => {
        :userName => plan_package[:userName],
        :script   => plan_package[:scripts][:script] 
      },
      :nodes => [],
      :edges => [],
      :node_groups => []
    }

    translate_status(plan_package[:status], result)
    
    [:optimized, :unoptimized].each do |plan|
      translate_plan(plan.to_s, plan_package[plan][:plan], result)
    end

    idx = 0
    nodeGroups.each do |ngId, ng|
      node = {
        :id => ngId,
        :child => idx.to_s        
      }

      node_group = {
        :id         => idx.to_s,
        :children   => ng[:children],
        :properties => ng[:properties],
        :status     => ng[:status]
      }

      result[:nodes] << node
      result[:node_groups] << node_group
      
      idx += 1
    end

    [:optimized, :unoptimized].each do |s|
      node = {
        :id => s,
        :child => idx.to_s
      }


      children = plan_package[s][:plan].map do |node|
        s.to_s+"-"+node.last[:mapReduce][:jobId]
      end.sort.uniq
      
      node_group = {
        :id => idx.to_s,
        :children => children
      }

      result[:nodes] << node
      result[:node_groups] << node_group
      idx += 1
    end

    @json = JSON.generate(result)
  end

  def translate_status status, r
    r[:status] = {
      :heartbeatTime => status[:heartbeatTime],
      :progress      => status[:progress],
      :startTime     => status[:startTime],
      :endTime       => status[:endTime],
      :statusText    => status[:statusText]
    }

    status[:jobStatusMap].each do |jobId, jobStatus|
      scope = jobStatus[:scope]
      mapProgress = (jobStatus[:mapProgress] ? jobStatus[:mapProgress].to_f : 0)
      reduceProgress = (jobStatus[:reduceProgress] ? jobStatus[:reduceProgress].to_f : 0)
      statusText = ""
      if jobStatus[:isComplete]
        if jobStatus[:isSuccessful]
          statusText = "finished"
        else
          statusText = "failed"
        end
      else
        statusText = "running"
      end
      
      newStatus = {
        :progress => (mapProgress + reduceProgress)/2,
        :startTime => jobStatus[:startTime],
        :endTime => jobStatus[:finishTime],
        :statusText => statusText
      }
      ["unoptimized", "optimized"].each do |s|
        nodeGroups[s+"-"+scope][:properties] = jobStatus
        nodeGroups[s+"-"+scope][:status] = newStatus
      end      
    end
    
  end
  
  def translate_join join, properties
    by = join[:expression].map do |name, value|
      {:alias => name, :fields => value[:fields]}
    end
    
    properties[:join] = {
      :strategy => join[:strategy],
      :type     => join[:type],
      :by       => by     
    }   
  end

  def translate_node prefix, node, r
    rNode = {
      :type => "PigNode",
      :id   => prefix + "-" + node[:uid],
    }
    properties = {
      :alias     => node[:alias],
      :schema    => node[:schema],
      :step_type => node[:mapReduce][:stepType],
      :location  => node[:location],
      :operation => node[:operator].gsub("LO", "").upcase,
    }

    edges = node[:successors].map do |succ|
      {:u => prefix+"-"+node[:uid], :v => prefix+"-"+succ}
    end

    r[:edges] += edges

    if node[:storageFunction]
      properties[:storage_function] = node[:storageFunction]
    end

    if node[:storageLocation]
      properties[:storage_location] = node[:storageLocation]
    end

    if node[:group]
      properties[:operation] = "GROUP"
      translate_join(node[:group], properties) 
    end

    if node[:join]
      translate_join(node[:join], properties)
    end
    
    rNode[:properties] = properties
    r[:nodes] << rNode

    scope = prefix + "-" + node[:mapReduce][:jobId]
    nodeGroups[scope][:children] ||= []
    nodeGroups[scope][:children] <<  prefix+"-"+node[:uid]
  end

  def translate_plan prefix, plan, r
    plan.each do |k,v|
      translate_node(prefix, v, r)
    end  
  end
end

data = JSON.parse(File.read(ARGV[0]), {:symbolize_names => true})
json = P2jTranslator.new(data).json

graph = Lipstick::Graph.from_json(json)

puts graph.to_json