gistfile1.txt

import ujson as json
import matplotlib.pyplot as plt
from moztelemetry import get_pings, get_pings_properties, get_one_ping_per_client
import datetime

def whatTime():
    print datetime.datetime.now()

## Get the v4 main pings for Beta users during 07/15 ... 08/31 I need
## all the pings hence fraction =1 since random sampling at this
## stage, would also imply random sampling for a given clients pings
## too.
pings = get_pings(sc, schema='v4', app="Firefox", channel="beta", doc_type="main"
                  , submission_date = ("20150701","20150915"),fraction=1)

## This function is used below to extract column names I used this
## since i got an error when using extract fields function

def getPath(p, pa,miss=None):
    try:
        if p is None:
            p = {}
        for ff in pa.split("/"):
            p = p.get(ff)
            if p is None:
                p = {}
        if p is None or p == {}:
            return miss
        else:
            return p
    except:
        return miss

## The filter used for returning a sample
def getSample(p):
    try:
        return (getPath(p,"environment/system/os/name","missing") == u'Windows_NT' and getPath(p,"meta/sampleId",1000.0) < 30.0)
    except:
        return False

## The columns required
def SelectColumns(p):
    try:
        m = { 'clientId'                 : getPath(p,"clientId","missing"),
              'creationDate'             : getPath(p,"environment/profile/creationDate",-1),
              'isdefault'                : getPath(p,"environment/settings/isDefaultBrowser",-1),
              "name"                     : getPath(p,"environment/system/os/name","missing"),
              "osversion"                : getPath(p,"environment/system/os/version","missing"),
              "install"                  : getPath(p,"environment/system/os/installYear",-1),
              "fxversion"                : getPath(p,"application/version","missing"),
              "sessionStartDate"         : getPath(p,"payload/info/sessionStartDate"),
              "subsessionStartDate"      : getPath(p,"payload/info/subsessionStartDate"),
              "timezoneOffset"           : getPath(p,"payload/info/timezoneOffset",-1),
              "profileSubsessionCounter" : getPath(p,"payload/info/profileSubsessionCounter",-1),
              "subsessionCounter"        : getPath(p,"payload/info/subsessionCounter",-1),
              "subsessionLength"         : getPath(p,"payload/info/subsessionLength",-1),
              "activeTicks"              : getPath(p,"payload/simpleMeasurements/activeTicks",-1),
              "totalTime"                : getPath(p,"payload/simpleMeasurements/totalTime",-1),
              "searches"                 : getPath(p,"payload/keyedHistograms/SEARCH_COUNTS",{}),
              "sampleId"                 : getPath(p,"meta/sampleId",-1.0)
        }
        return m
    except:
        return {}

whatTime()

pings2 = pings.filter(getSample)

whatTime()


## lets save this for tomorrow
## makes sense to map the values after filtering
## why run transformation code on all the data
## when i only need 30% of it?
## I need the coalesce because o/w i have thousands of files which take time just to read
pings3 = pings2.map(SelectColumns)
pings3.map(lambda x: (x['clientId'],json.dumps(x))).coalesce(400).saveAsHadoopFile(path="s3n://net-mozaws-prod-us-west-2-pipeline-analysis/sguha/win10checks3pct30-a0"
                        ,outputFormatClass="org.apache.hadoop.mapred.SequenceFileOutputFormat"
                        ,keyClass="org.apache.hadoop.io.Text"
                        ,valueClass="org.apache.hadoop.io.Text")