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 



15
16
17
18
19
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 15

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

#emit_presto_query ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 45

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

#run ⇒ Object 



36
37
38
39
40
41
42
43
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 36

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 



32
33
34
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 32

def shutdown
  Thread.kill(@thread)
end

#start ⇒ Object 



21
22
23
24
25
26
27
28
29
30
 
# File 'lib/fluent/plugin/in_presto_query.rb', line 21

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