class Fluent::BigObjectOutput_AVRO
Attributes
tables[RW]
Public Class Methods
new()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_bigobject_avro.rb, line 109 def initialize super require 'avro' log.info("bigobject_avro initialize") end
Public Instance Methods
configure(conf)
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_bigobject_avro.rb, line 115 def configure(conf) super if remove_tag_prefix = conf['remove_tag_prefix'] @remove_tag_prefix = Regexp.new('^' + Regexp.escape(remove_tag_prefix)) end @tables = [] @default_table = nil conf.elements.select { |e| e.name == 'table' }.each { |e| te = TableElement.new(log, @bigobject_hostname, @bigobject_port) te.configure(e) @tables << te } # @tables.each {|t| puts t.to_s} end
emit(tag, es, chain)
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_bigobject_avro.rb, line 173 def emit(tag, es, chain) super(tag, es, chain, format_tag(tag)) end
format(tag, time, record)
click to toggle source
This method is called when an event reaches to Fluentd.
# File lib/fluent/plugin/out_bigobject_avro.rb, line 146 def format(tag, time, record) [tag, time, record].to_msgpack end
format_tag(tag)
click to toggle source
# File lib/fluent/plugin/out_bigobject_avro.rb, line 165 def format_tag(tag) if @remove_tag_prefix tag.gsub(@remove_tag_prefix, '') else tag end end
shutdown()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_bigobject_avro.rb, line 141 def shutdown super end
start()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_bigobject_avro.rb, line 136 def start super log.info("bigobject_avro start") end
write(chunk)
click to toggle source
This method is called every flush interval. Write the buffer chunk to files or databases here. ‘chunk’ is a buffer chunk that includes multiple formatted events.
# File lib/fluent/plugin/out_bigobject_avro.rb, line 153 def write(chunk) unknownChunks = [] @tables.each { |table| if table.mpattern.match(chunk.key) return table.send_binary(chunk) end } log.warn("unknown chunk #{chunk.key}") end