Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Aktsk-Fluentd+MongoDB

Yusuke
March 25, 2013
810

 Aktsk-Fluentd+MongoDB

Yusuke

March 25, 2013
Tweet

Transcript

  1. ઃఆ [APαʔόଆ] <source>    type  tail_ex    format  tsv  

     path  /media/ephemeral0/bigdata/payment.log    keys  amount,item_num,(...লུ)    int  item_num    float  amount    tag  app.payment    pos_file  /var/log/td-­‐agent/payment.log.pos </source>
  2. ઃఆ [APαʔόଆ] <source>    type  tail    format  tsv  

     path  /media/ephemeral0/bigdata/access.log    keys  log_time,url,...(লུ)    tag  app.access    pos_file  /var/log/td-­‐agent/access.log.pos </source>   <match  app.access>    type  forward    flush_interval  5s    <server>        host  xxxxxxx        port  xxxx    </server> </match> [ूܭαʔόଆ] <match  app.access>    type  extract_query_params    add_tag_suffix  .extracted    key            url    only          ab    add_path  page </match> <match  app.access.extracted>    type  mongo    host  localhost    database  xxxxxxxx    collection  log.accesses    capped    capped_size  100m    flush_interval  10s </match>
  3. • Capped-CollectionʹΞΫηεϩάΛྲྀ͜͠Μ ͰɺΧʔιϧΛॱ࣍ॲཧͯ͠Page-ViewίϨ ΫγϣϯΛߋ৽ • UU͸countͰɺPV͸pvଐੑͷ૯ܭͰදݱ Capped-Collection _id:    

             #{yyyymmddhh}  +  ":"  +  #{page}  +  ":"  +  #{user_id} date_hour:  Time page:            String user_id:      Integer pv:                Integer
  4. #  ӬଓԽ͓͍ͯͨ͠࠷ޙͷߋ৽࣌ؒ processed_time  =  Tmp::PagePosition.first.processed_time loop  do    #  Χʔιϧ͕ด͡ΒΕͨ࣌ʹ࠶։͢Δ

       cursor  =  Mongo::Cursor.new('log.accesses',  selector:  {'log_time'  =>  {'$gt'  =>  processed_time}},                                                            timeout:  false,  tailable:  true,  order:  [['$natural',  1]])    loop  do        break  if  cursor.closed?        begin            document  =  cursor.next_document            unless  document  then                sleep(0.1)                next            end              <ϖʔδϏϡʔίϨΫγϣϯͷߋ৽ॲཧ>              if  document["log_time"]  >  processed_time  then                processed_time  =  log_time                Tmp::PagePosition.update_processed_time(processed_time)            end        rescue  =>  e            <Τϥʔग़ྗ>            cursor.close            break        end    end end