Class: Fluent::PrestoQueryInput

Inherits:
Input
  • Object
show all
Defined in:
lib/fluent/plugin/in_presto_query.rb

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object 



20
21
22
23
24
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 20

def configure(conf)
  require 'presto-client'
  require 'parse-cron'
  super
end

#emit_presto_query ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 50

def emit_presto_query
  begin
    log.info "sql [#{@sql}]"
    records = exec_query(@sql)
    records.each do |record|
      router.emit @tag, Fluent::Engine.now, record
    end
  rescue => e
    log.error e
  end
end

#run ⇒ Object 



41
42
43
44
45
46
47
48
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 41

def run
  loop do
    secs = @cron_parser.next(Time.now) - Time.now
    log.info "next query at #{@cron_parser.next(Time.now)}. Sleep #{secs}seconds."
    sleep secs
    Thread.new(&method(:emit_presto_query))
  end
end

#shutdown ⇒ Object 



37
38
39
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 37

def shutdown
  Thread.kill(@thread)
end

#start ⇒ Object 



26
27
28
29
30
31
32
33
34
35
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 26

def start
  @cron_parser = CronParser.new(@cron)
  @client = Presto::Client.new(
    server: "#{@host}:#{@port}",
    catalog: @catalog,
    user: @user,
    schema: @schema
  )
  @thread = Thread.new(&method(:run))
end