python spark example

這是年初寫的一個job，用於對api非法訪問（大量403）進行統計，並有其他后續手段。寫的比較通俗易懂，做個sample記錄下

數據源是kafka stream，實時計算。規則是mysql配置的，簡單說就是1分鍾內超過多少次403就記錄下來

import json
import logging
from datetime import datetime

import MySQLdb
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

logger = logging.getLogger()
hdlr = logging.FileHandler('nginx_log_stats.log')
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.ERROR)


def is_in_one_minute(nginx_timestamp):
    """
    :param nginx_time: "timestamp": "16/Feb/2017:08:23:59 +0000"
    :return:
    """
    now = datetime.now()
    nginx_datetime = datetime.strptime(nginx_timestamp.split('+')[0].strip(),
                                       '%d/%b/%Y:%H:%M:%S')
    return (now - nginx_datetime).seconds <= 60 if now > nginx_datetime else (nginx_datetime - now).seconds <= 60



# save to mysql
def saveToMysql(partition):
    host = "..."
    user = "..."
    password = "..."
    db_name = "..._monitor"
    db = MySQLdb.connect(host, user, password, db_name, charset='utf8')
    db.autocommit(True)
    cursor = db.cursor()
    for d1ct in partition:
        sql = r"""INSERT INTO `security_suspect_request` (`domain`, `api`, `code`, `ip`, `access_count`) VALUES ('{domain}', '{api}', '{code}', '{ip}', {access_count})""".format(
            domain=d1ct['domain'], api=d1ct['path'], code=d1ct['response'], ip=d1ct['ip'],
            access_count=d1ct['count'])
        cursor.execute(sql)
    db.close()


def dictfetchall(cursor):
    "Return all rows from a cursor as a dict"
    columns = [col[0] for col in cursor.description]
    return [
        dict(zip(columns, row))
        for row in cursor.fetchall()
        ]


def filterPolicy(log):
    '''
{
  "path": "/var/log/nginx/webapi..../access-log",
  "host": "ip-10-...",
  "clientip": "10....",
  "timestamp": "16/Feb/2017:08:23:59 +0000",
  "domain": "...com",
  "verb": "POST",
  "request_path": "/video/upload",
  "request_param": "sig=b400fdce...&userId=...",
  "httpversion": "1.1",
  "response": "403",
  "bytes": "0",
  "agent": "Dalvik/1.6.0 (Linux; U; Android 4.4.4; SM-T561 Build/KTU84P)",
  "response_time": "0.110",
  "topic": "nginx"
}
    '''
    # true save . false ignore
    true_flag = 0
    this = json.loads(log[1])
    # filter time
    if not is_in_one_minute(this['timestamp']):
        return False
    # filter condition
    for policy in filterVar.value:
        if policy['domain'] == 'all' or ('domain' in this.keys() and this['domain'] == policy['domain']):
            if policy['api'] == 'all' or ('request_path' in this.keys() and this['request_path'] == policy['api']):
                if 'response' in this.keys() and this['response'] == str(policy['code']):
                    true_flag += 1

    return True if true_flag else False


def countMap(log):
    import json, re
    this = json.loads(log[1])
    key = this.get('domain', "") + "--" + re.sub(r'\/\d+$', r'',
                                                 this.get('request_path', "") + "--" + this.get(
                                                     'clientip') + "--" + this.get('response'))
    value = {'count': 1}
    return key, value


def countReduce(prev, cur):
    cur['count'] = cur['count'] + prev['count']
    return cur


def output(tup1e):
    """
    a touple (key, value)
    """
    tup1e[1]['domain'], tup1e[1]['path'], tup1e[1]['ip'], tup1e[1]['response'] = tup1e[0].split('--')
    return tup1e[1]


def youAreUnderArrest(d1ct):
    mylimit = None
    for row in filterVar.value:
        if row['domain'] == 'all' or row['domain'] == d1ct['domain']:
            if row['api'] == 'all' or row['api'] == d1ct['path']:
                if row['code'] == int(d1ct['response']):
                    mylimit = row['limit']

    return False if mylimit is None else d1ct['count'] >= mylimit


if __name__ == "__main__":
    host = "..."
    user = "..."
    password = "..."
    db_name = "..._monitor"
    db = MySQLdb.connect(host, user, password, db_name, charset='utf8')
    db.autocommit(True)
    cur = db.cursor()
    try:
        # for now only support 1 row
        cur.execute(r"""SELECT * FROM security_anti_hacker_policy""")
        filter_option = dictfetchall(cur)
    finally:
        db.close()

    topic = 'nginx.log'
    zkQuorum = '...:2181,...:2181,...:2181'
    conf = (SparkConf()
            .setMaster("spark://...:7077")
            .setAppName("anti_hacker_stats")
            .set("spark.driver.memory", "1g")
            .set("spark.executor.memory", "1g")
            .set("spark.cores.max", 2))
    sc = SparkContext(conf=conf)
    # broadcast variable for share
    filterVar = sc.broadcast(filter_option)
    ssc = StreamingContext(sc, 60)
    kvs = KafkaUtils.createStream(ssc, zkQuorum, "anti-hacker", {topic: 1},
                                  {"auto.offset.reset": 'largest'})
    lines = kvs.filter(filterPolicy).map(countMap).reduceByKey(countReduce).map(output).filter(youAreUnderArrest)
    lines.foreachRDD(lambda rdd: rdd.foreachPartition(saveToMysql))
    # lines.saveAsTextFiles('test')
    # lines = kvs.filter(filterPolicy)
    # lines.pprint()
    ssc.start()
    ssc.awaitTermination()

python寫spark需要在spark服務器上用pyspark執行，調試很不方便，更建議用scala，另有example

幾個重點：

1. 因為spark天然就是分布式的，所以每個rdd可以認為就是在在不同的機器上，是不能共享jdbc connection的，需要各寫各的
2. 因為上面那個原因，如果需要共享數據呢？很直觀，就是150行那句sc.broadcast，將共享數據廣播給各個rdd
3. 數據格式很重要，你必須了解數據源里的格式