spark1.6学习（四)——计算pv和uv的例子

小灰灰 2022-04-13 10:10 145阅读 0赞

本文主要介绍如何通过spark进行pv和uv的计算。一般我们经常会计算pv和uv，那么我们计算pv和uv的时候是不是性能最优的呢？

好，我们开始看例子：

首先看一下数据：

{"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_cccc","timestamp":1543309410741,"device":null,"ip":null,"bucket":1,"data":{"templateName":"votePost","appType":1,"sendNum":1}}
    {"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_rvm-lhpa2gPc7QBs","timestamp":1543309410741,"device":null,"ip":null,"bucket":4,"data":{"templateName":"dailySignPush","appType":3,"sendNum":1}}
    {"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_rvm-eeee","timestamp":1543309410741,"device":null,"ip":null,"bucket":5,"data":{"templateName":"replyPost","appType":2,"sendNum":1}}
    {"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_rvm-xxxxx","timestamp":1543309410741,"device":null,"ip":null,"bucket":4,"data":{"templateName":"dailySignPush","appType":3,"sendNum":1}}
    {"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_rvm-pppeeee","timestamp":1543309410741,"device":null,"ip":null,"bucket":4,"data":{"templateName":"dailySignPush","appType":3,"sendNum":1}}
    {"flag":"sendTemplateMessage","actionType":"success","from":"sendTemplateMessage","openId":"otU065OELPd_rvm-lhpa2gPc7QBs","timestamp":1543309410741,"device":null,"ip":null,"bucket":4,"data":{"templateName":"dailySignPush","appType":3,"sendNum":1}}

然后我们按行读取数据，读取后，我们需要算出不同bucket中不同openid的sendNum的pv和uv，其中pv为sendNum的总和，uv为不重复的openId数。

#!/usr/bin/python
    # -*- coding: UTF-8 -*-
    from pyspark import SparkContext, SparkConf
    import json
    
    
    def parseJson(log_line):
        json_dic = json.loads(log_line)
        print json_dic
        return (json_dic["flag"], json_dic["actionType"], json_dic["data"]["appType"], json_dic["data"]["templateName"],
                json_dic["bucket"], json_dic["openId"], json_dic["data"]["sendNum"])
    
    
    def fileterRdd(line):
        a, b, c, d, e, f, g = line
        if a == 'sendTemplateMessage' and b == 'success':
            return True
        else:
            return False
    
    
    def splitRdd(line):
        a, b, c, d, e, f, g = line
        return ((c, d, e, f), g)
    
    
    def transformRdd(line):
        (c, d, e, f), g = line
        return ((c, d, e), (f, g, 1))
    
    
    def caculateRes(line1, line2):
        f, g, k = line1
        f2, g2, k2 = line2
        return (f, g + g2, k + k2)
    
    
    def main():
        logFile = "/user/root/spark/sparkstudy02.txt"
        master = 'yarn-client'
        appName = 'Simple App spark study02'
        conf = SparkConf().setAppName(appName).setMaster(master)
        sc = SparkContext(conf=conf)
        logData = sc.textFile(logFile)
        logStage1 = logData.map(lambda x: parseJson(x))
        logStage2 = logStage1.filter(lambda x: fileterRdd(x))
        logStage3 = logStage2.map(lambda x: splitRdd(x))
        logStage4 = logStage3.reduceByKey(lambda x, y: x + y)
        logStage5 = logStage4.map(lambda x: transformRdd(x))
        logStage6 = logStage5.reduceByKey(lambda x, y: caculateRes(x, y))
        print("============================")
        for item in logStage1.collect():
            print(item)
        print("============================")
        for item in logStage2.collect():
            print(item)
        print("============================")
        for item in logStage3.collect():
            print(item)
        print("============================")
        for item in logStage4.collect():
            print(item)
        print("============================")
        for item in logStage5.collect():
            print(item)
        print("============================")
        for item in logStage6.collect():
            print(item)
    
    
    if __name__ == '__main__':
        main()

运行结果如下：

((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-lhpa2gPc7QBs', 2, 1))
    ((1, u'votePost', 1), (u'otU065OELPd_cccc', 1, 1))
    ((2, u'replyPost', 5), (u'otU065OELPd_rvm-eeee', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-pppeeee', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-xxxxx', 1, 1))
    ============================
    ((1, u'votePost', 1), (u'otU065OELPd_cccc', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-lhpa2gPc7QBs', 4, 3))
    ((2, u'replyPost', 5), (u'otU065OELPd_rvm-eeee', 1, 1))
    [root@cdh1 demo-simple]#
    [root@cdh1 demo-simple]#
    [root@cdh1 demo-simple]#
    [root@cdh1 demo-simple]# bash sparkstudy02.sh
    ============================
    (u'sendTemplateMessage', u'success', 1, u'votePost', 1, u'otU065OELPd_cccc', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs', 1)
    (u'sendTemplateMessage', u'success', 2, u'replyPost', 5, u'otU065OELPd_rvm-eeee', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-xxxxx', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-pppeeee', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs', 1)
    ============================
    (u'sendTemplateMessage', u'success', 1, u'votePost', 1, u'otU065OELPd_cccc', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs', 1)
    (u'sendTemplateMessage', u'success', 2, u'replyPost', 5, u'otU065OELPd_rvm-eeee', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-xxxxx', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-pppeeee', 1)
    (u'sendTemplateMessage', u'success', 3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs', 1)
    ============================
    ((1, u'votePost', 1, u'otU065OELPd_cccc'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs'), 1)
    ((2, u'replyPost', 5, u'otU065OELPd_rvm-eeee'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-xxxxx'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-pppeeee'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs'), 1)
    ============================
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-lhpa2gPc7QBs'), 2)
    ((1, u'votePost', 1, u'otU065OELPd_cccc'), 1)
    ((2, u'replyPost', 5, u'otU065OELPd_rvm-eeee'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-pppeeee'), 1)
    ((3, u'dailySignPush', 4, u'otU065OELPd_rvm-xxxxx'), 1)
    ============================
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-lhpa2gPc7QBs', 2, 1))
    ((1, u'votePost', 1), (u'otU065OELPd_cccc', 1, 1))
    ((2, u'replyPost', 5), (u'otU065OELPd_rvm-eeee', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-pppeeee', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-xxxxx', 1, 1))
    ============================
    ((1, u'votePost', 1), (u'otU065OELPd_cccc', 1, 1))
    ((3, u'dailySignPush', 4), (u'otU065OELPd_rvm-lhpa2gPc7QBs', 4, 3))
    ((2, u'replyPost', 5), (u'otU065OELPd_rvm-eeee', 1, 1))

最后我们进行简要的分析：

注意，还有一种方法是通过groupbykey的方式，同时可以通过distinct()操作过滤掉重复的数据从而实现uv，但是这里没有使用这种方法。

原因有如下两个：

1、reduceByKey相比groupByKey在计算效率上会更好一些。

Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or aggregateByKey will provide much better performance.

2、如果采用两次计算，一次使用所有数据，一次使用distinct()的数据，那么最后汇总的时候如何处理呢？

同时，在处理时也会出现同一个rdd使用多次的现象，虽然我们可以使用cache把rdd暂时保存在内存中，但是我们应该尽量去使用能够一次的到pv和uv的方法。