Python 批量写入 Elasticsearch 脚本

#!/usr/bin/env pypy
#coding:utf-8

import re
import sys
import time
import datetime
import logging
from elasticsearch import Elasticsearch
from elasticsearch import helpers
from elasticsearch import ConnectionTimeout

es = Elasticsearch(['192.168.0.2', '192.168.0.3'], sniff_on_start=True, sniff_on_connection_fail=True, max_retries=3, retry_on_timeout=True)
logging.basicConfig()
logging.getLogger('elasticsearch').setLevel(logging.WARN)
logging.getLogger('urllib3').setLevel(logging.WARN)

def parse_www(logline):
	try:
		time_local, request, http_user_agent, staTus, remote_addr, http_referer, request_time, body_bytes_sent, http_x_forwarded_proto, http_x_forwarded_for, http_host, http_cookie, upstream_response_time = logline.split('`')
		try:
			upstream_response_time = float(upstream_response_time)
		except:
			upstream_response_time = None

		method, uri, verb = request.split(' ')
		arg = {}
		try:
			url_path, url_args = uri.split('?')
			for args in url_args.split('&'):
				k, v = args.split('=')
				arg[k] = v
		except:
			url_path = uri

		# Why %z do not implement?
	        date = datetime.datetime.strptime(time_local, '[%d/%b/%Y:%H:%M:%S +0800]')
		ret = {
			"@timestamp": date.strftime('%FT%T+0800'),
			"host": "127.0.0.1",
			"method": method.lstrip('"'),
			"url_path": url_path,
			"url_args": arg,
			"verb": verb.rstrip('"'),
			"http_user_agent": http_user_agent,
			"status": int(staTus),
			"remote_addr": remote_addr.strip('[]'),
			"http_referer": http_referer,
			"request_time": float(request_time),
			"body_bytes_sent": int(body_bytes_sent),
			"http_x_forwarded_proto": http_x_forwarded_proto,
			"http_x_forwarded_for": http_x_forwarded_for,
			"http_host": http_host,
			"http_cookie": http_cookie,
			"upstream_response_time": upstream_response_time
		}
		return {"_index":"logstash-mweibo-www-"+date.strftime('%Y.%m.%d'), "_type":"nginx","_source":ret}
	except:
		return {"_index":"logstash-mweibo-www-"+datetime.datetime.now().strftime('%Y.%m.%d'), "_type":"nginx","_source":{"message":logline}}

def get_log():
    start_time = time.time()
    log_buffer = []
    while True:
        try:
            line = sys.stdin.readline()
        except:
            break
        if not line:
            helpers.bulk(es, log_buffer)
            del log_buffer[0:len(log_buffer)]
            break

        if line:
            ret = parse_www(line.rstrip())
            log_buffer.append(ret)
            while ( len(log_buffer) > 2000 and len(log_buffer) % 2000 == 0 ):
                try:
                    helpers.bulk(es, log_buffer)
                except ConnectionTimeout:
                    print("try again")
                    continue
                del log_buffer[0:len(log_buffer)]
                break

        else:
            if (time.time() - startime > timeout ):
                helpers.bulk(es, log_buffer)
                start_time = time.time()
                del log_buffer[0:len(log_buffer)]
            time.sleep(1)

if __name__ == '__main__':
    get_log()