squid 日志分析 - hadoop hive

 #!/usr/bin/env python

#-*-coding:UTF-8-*-
"""
@Item   :  Hadoop analysis squid log
@Author :  Villiam Sheng
@Group  :  Linux Group
@Date   :  2012-09-13
@Funtion:
            Use hadoop squid processing log ......................
"""
import os,re,threading,tarfile,shutil,time,sys,datetime
import sys,traceback
sys.path.append('/usr/local/hive/lib/py')
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from hive_service import ThriftHive
from hive_service.ttypes import HiveServerException
 
 
class split_log(threading.Thread):
    def __init__(self,files,lock):
        threading.Thread.__init__(self)
        self.files  = files
        self.path = '/data1/squid_log'
        self.lock = lock
 
    def log(self,info):
        files = open('%s/split_log'%self.path,'a')
        try:
            files.write(info)
        except IOError:
            files.close()
        files.close()
        size = os.path.getsize('%s/split_log'%self.path) / 1024
        if size >= 1024:
            fp = open('%s/split_log'%self.path,'w')
            fp.write('')
            fp.close()
 
    def run(self):
        try:
            if not os.path.isdir(self.files):
                tar = tarfile.open('%s/%s'%(self.path,self.files))
                for name in tar.getnames():
                    tar.extract(name,'/%s/%s/'%(self.path,self.files.split('.tgz')[0]))
                tar.close()
                self.log('%s file:%s run.info:tar done \n'%(time.ctime(),self.files))
        except Exception,e:
            self.log('%s files:%s run.info:%s \n'%(time.ctime(),self.files,e))
        try:
            fp = open('%s/%s/access.log.0'%(self.path,self.files.split('.tgz')[0]),'r')
            for i in fp.readlines():
                b = i.split()
                fps = open('%s/%s/tmp.log'%(self.path,self.files.split('.tgz')[0]),'a+')
                fps.write("%s\t%s\t%s\t%s\t%s\n" %(datetime.datetime.strptime(b[3].split('[')[1],'%d/%b/%Y:%H:%M:%S').strftime('%Y%m%d%H%M%S'),b[0],i.split('/')[4],b[-5],b[-1]))
            fps.close()
            fp.close()
        except Exception,e:
             self.log('%s file:%s inert.info:insert done \n'%(time.ctime(),self.files.split('.tgz')[0]))
        try:
            asql="load data local inpath \'%s/%s/tmp.log\' overwrite into table squid.squid_tmp partition(pt_ip = \'%s\')" %(self.path,self.files.split('.tgz')[0],self.files.split('.tgz')[0])
            self.lock.acquire()
            transport = TSocket.TSocket('127.0.0.1',10000)
            transport = TTransport.TBufferedTransport(transport)
            protocol = TBinaryProtocol.TBinaryProtocol(transport)
            client = ThriftHive.Client(protocol)
            transport.open()
            try:
                client.execute(asql)
            except:
                pass
            transport.close()
            self.lock.release()
            os.remove ('%s/%s'%(self.path,self.files))
            shutil.rmtree('%s/%s' %(self.path,self.files.split('.tgz')[0]))
        except:
            self.log('Exception error : %s '%traceback.print_exc())
   
 
def work():
    ip_list = re.compile('^(([1-9]|([1-9]\d)|(1\d\d)|(2([0-4]\d|5[0-5])))\.)(([1-9]|([1-9]\d)|(1\d\d)|(2([0-4]\d|5[0-5])))\.){2}([1-9]|([1-9]\d)|(1\d\d)|(2([0-4]\d|5[0-5])))$')
    lock = threading.RLock()
    for i in os.listdir('/data1/squid_log/'):
        if not ip_list.match(i.split('.tgz')[0]):
            continue
        st = split_log(i,lock)
        st.start()
    while threading.active_count() != 1:
        time.sleep(1)
 
    transport = TSocket.TSocket('127.0.0.1',10000)
    transport = TTransport.TBufferedTransport(transport)
    protocol = TBinaryProtocol.TBinaryProtocol(transport)
    client = ThriftHive.Client(protocol)
    transport.open()
    client.execute("use squid")
    client.execute("set hive.exec.dynamic.partition=true")
    client.execute("set hive.exec.dynamic.partition.mode=nonstrict")
    sql = "insert into table squid_log partition(pt_ip, pt_dt) select visittime, clientip, visitdom, visiturl, visitstat, pt_ip, substr(visittime, 1, 8) as pt_dt from squid_tmp"
    client.execute(sql)
    transport.close()
 
if __name__ == "__main__":
    try:
        pid = os.fork()
        if pid > 0 :
            sys.exit(0)
        os.setsid()
        os.chdir('/')
        sys.stdin = open("/dev/null","r+")
        sys.stdout = os.dup(sys.stdin.fileno())
        sys.stderr = os.dup(sys.stdin.fileno())
        while True:
            work()
            time.sleep(300)
        work()
    except:
        pass

本文转自 swq499809608 51CTO博客,原文链接:http://blog.51cto.com/swq499809608/1142212
上一篇:人生苦短,开发用云-如何优雅完成程序员的侠客梦


下一篇:Docker容器可视化监控中心搭建