code.2.py

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import plotly.plotly as py
sc.setJobGroup(groupId = "sguha",description = "interactive")


frame = sqlContext.read.load("s3://telemetry-parquet/longitudinal/v20160321")
first = frame.filter("normalized_channel = 'beta'").limit(5).rdd.collect()

profilePassed = {'total':sc.accumulator(0),'hasSubmissions':sc.accumulator(0),'didPass':sc.accumulator(0),'nulle10s':sc.accumulator(0)}
def smry(p):
    clientid                          = p.client_id
    submissiondates                   = [ x[0:10] for x in p.submission_date ]
    slength                           = p.subsession_length or [0]*len(submissiondates)
    buildid                           = [x.build_id[:8] for x  in p.build] or ['20150101'] * len(submissiondates)
    if p.subprocess_crashes_with_dump is None:
        subprocess_crashes_with_dump  = {}
    else:
        subprocess_crashes_with_dump  = p.subprocess_crashes_with_dump
    crashplugin                       = subprocess_crashes_with_dump.get("content",[0]*len(submissiondates))
    crashcontent                      = subprocess_crashes_with_dump.get("plugin",[0]*len(submissiondates))
    version                           = [x.version for x in p.build]
    e10s                              = [x.e10s_enabled for x in p.settings]
    
    ## We want to collect
    ## during the period 20160305-20160311
    ## for beta, version 46.0
    ## profile id, e10s, total session time, total contentcrash, total plugincrash
    ## and profile must be on beta 46
    addOnce = True
    plugincrashes  = {True:0, False:0}
    contentcrashes = {True:0, False:0}
    totaltime      = {True:0, False:0}
    profilePassed['total'].add(1)
    if len(submissiondates)>0:
        profilePassed['hasSubmissions'].add(1)
        for bd, ver, pc,cc, tt,e10 in zip(buildid,version,crashplugin,crashcontent,slength,e10s):
            if  ver=="46.0": #bd >= "20160307" and bd<="20160315" and
                if addOnce:
                    profilePassed['didPass'].add(1)
                    addOnce = False
                plugincrashes[e10] += pc
                contentcrashes[e10] += cc
                totaltime[e10] += tt
        if not addOnce and any(e10s):
            yield ",".join([str(x) for x in [clientid, True, totaltime[True],contentcrashes[True],plugincrashes[True]]])
        if not addOnce and not all(e10s):
            yield ",".join([str(x) for x in [clientid, False, totaltime[False],contentcrashes[False],plugincrashes[False]]])


sf = frame.filter("normalized_channel = 'beta'").select("client_id","submission_date","subsession_length","subprocess_crashes_with_dump","settings","build")
sf3=sf.flatMap(smry).coalesce(100)

sf3.saveAsTextFile("s3://telemetry-test-bucket/sguhatmp/crashhours14")
"Total Profiles = {}, PassedWithSubmission = {} , isOfInterest = {} ".format(profilePassed['total'].value,
                                                                                             profilePassed['hasSubmissions'].value,
                                                                                             profilePassed['didPass'].value)



## Different versions present
from operator import add
profilePassed = {'total':sc.accumulator(0),'hasSubmissions':sc.accumulator(0),'didPass':sc.accumulator(0),'nulle10s':sc.accumulator(0)}
def smry2(p):
    submissiondates                   = [ x[0:10] for x in p.submission_date ]
    buildid                           = [x.build_id[:8] for x  in p.build] or ['20150101'] * len(submissiondates)
    version                           = [x.version for x in p.build]
    profilePassed['total'].add(1)
    if len(submissiondates)>0:
        profilePassed['hasSubmissions'].add(1)
        for bd, ver in zip(buildid,version):
            return  ( (ver,bd),1) 

versions=sf.map(smry2).reduceByKey(add).collect()
print("Total Profiles = {}, PassedWithSubmission = {}".format(profilePassed['total'].value,
                                                        profilePassed['hasSubmissions'].value))
sorted(versions, key=lambda x: x[1])




def versionMap(p):
    version = [x.version for x  in p.build][0]
    return (version,1)
versions=sf.map(versionMap).reduceByKey(add).collect()


    
# x.fx_page_load_ms
#     first_paint

# subprocess_crashes_with_dump
#     subsession_length