Class: Fairy::PWC::PPostFilter

Inherits:

Fairy::PIOFilter

• Object
• Fairy::PFilter
• Fairy::PIOFilter
• Fairy::PWC::PPostFilter

Defined in:

lib/fairy/node/p-wc.rb

Constant Summary

ST_OUTPUT_FINISH =

:ST_OUTPUT_FINISH

IPADDR_REGEXP =

/::ffff:([0-9]+\.){3}[0-9]+|[0-9a-f]+:([0-9a-f]*:)[0-9a-f]*/

Constants inherited from Fairy::PIOFilter

Fairy::PIOFilter::ST_WAIT_IMPORT

Constants inherited from Fairy::PFilter

Fairy::PFilter::END_OF_STREAM, Fairy::PFilter::ST_ACTIVATE, Fairy::PFilter::ST_FINISH, Fairy::PFilter::ST_INIT

Instance Attribute Summary

Attributes inherited from Fairy::PFilter

#IGNORE_EXCEPTION, #id, #log_id, #ntask

Instance Method Summary

• #basic_start(&block) ⇒ Object
• #create_dir(path) ⇒ Object
• #gen_real_file_name ⇒ Object
• #hash_key(e) ⇒ Object
• #initialize(id, ntask, bjob, opt, vf) ⇒ PPostFilter constructor

  A new instance of PPostFilter.

• #input=(input) ⇒ Object

Methods inherited from Fairy::PFilter

#abort_running, #break_running, #each, #global_break, #global_break_from_other, #handle_exception, #key, #key=, #next, #no, #no=, #notice_status, #processor, #start, #start_export, #start_watch_status, #status=, #terminate, #terminate_proc

Constructor Details

#initialize(id, ntask, bjob, opt, vf) ⇒ PPostFilter

Returns a new instance of PPostFilter.

# File 'lib/fairy/node/p-wc.rb', line 162

def initialize(id, ntask, bjob, opt, vf)
	super
	@vfile = vf

	@buffering_policy = @opts[:buffering_policy]
	@buffering_policy ||= CONF.GROUP_BY_BUFFERING_POLICY

	@mod = @opts[:no_segment] 
	@mod ||= CONF.GROUP_BY_NO_SEGMENT

	mod = @opts[:hash_module]
	mod ||= CONF.GROUP_BY_HASH_MODULE
	require mod
	@hash_generator = Fairy::HValueGenerator.new(bjob.hash_seed)

	@hash_optimize = CONF.GROUP_BY_GROUPING_OPTIMIZE
	@hash_optimize = @opts[:grouping_optimize] if @opts.key?(:grouping_optimize)
end

Instance Method Details

#basic_start(&block) ⇒ Object

# File 'lib/fairy/node/p-wc.rb', line 190

def basic_start(&block)
	Log::debug(self, "START")
	output_uri = gen_real_file_name
	@vfile.set_real_file(no, output_uri)

	Log::debug(self, "write real file: #{output_uri}")
	begin
	  output_file = URI.parse(output_uri).path
	rescue
	  Log::debug_exception(self)
	  raise
	end

	unless File.exist?(File.dirname(output_file))
	  create_dir(File.dirname(output_file))
	end

	@key_value_buffer = 
	  eval("PGroupBy::#{@buffering_policy[:buffering_class]}").new(self, @buffering_policy)
#	if @hash_optimize
	  @hash_proc = eval("proc{|w| w}")
#	else
#	  @hash_proc = BBlock.new("|w| w", @context, self)
#	end

	@input.each do |e|
	  @key_value_buffer.push(e)
	  e = nil
	end

	File.open(output_file, "w") do |io|
	  Log::debug(self, "start write real file: #{output_uri}")
	  @key_value_buffer.each do |values|
	    io.puts [values.key, values.size].join(" ")
	  end
	  @key_value_buffer = nil
	  Log::debug(self, "finish write real file: #{output_uri}")
	end

	self.status = ST_OUTPUT_FINISH
end

#create_dir(path) ⇒ Object

# File 'lib/fairy/node/p-wc.rb', line 232

def create_dir(path)
	unless File.exist?(File.dirname(path))
	  create_dir(File.dirname(path))
	end
	begin
	  Dir.mkdir(path)
	rescue Errno::EEXIST
	  # 無視
	end
end

#gen_real_file_name ⇒ Object

# File 'lib/fairy/node/p-wc.rb', line 245

def gen_real_file_name
	host= processor.addr
	root = CONF.VF_ROOT
	prefix = CONF.VF_PREFIX
	base_name = @vfile.base_name
	no = @input.no
	

	if IPADDR_REGEXP =~ host
	  begin
	    host = Resolv.getname(host)
	  rescue
	    # ホスト名が分からない場合 は そのまま ipv6 アドレスにする
	    host = "[#{host}]"
	  end
	end
	
	format("file://#{host}#{root}/#{prefix}/#{base_name}-%03d", no)
end

#hash_key(e) ⇒ Object

# File 'lib/fairy/node/p-wc.rb', line 186

def hash_key(e)
	@hash_proc.yield(e)
end

#input=(input) ⇒ Object

# File 'lib/fairy/node/p-wc.rb', line 181

def input=(input)
	super
	start
end