## 12 Clean Spotify.r ## CLEAN { ## output an example, prior to cleaning cat("Here are some results, prior to cleaning the DT \n") print(head(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, 12)) print(tail(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, 12)) ## CLEANING cleanSpotify_ (DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, posix.to.date=FALSE, showWarnings=TRUE, verbose=TRUE) DT[, month := as.Date(strptime(month, format="%Y-%m-%d"))] # status # notifyAndEmail("Done cleaning DT. Setting key") ## Set key AFTER cleaining setcolorderpt(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, kCols_UCnDL) setkeyIfNot(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, kCols_UCnDL) # status # notify("done setting key") print(head(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, 12)) } ## product_integer is the underlying integer value of the factor. DT[, product_integer := as.integer(product)] ## Specifically, we do NOT want as.num.as.char ... we want the number underlying the factor ## -------------------------------------------------------------------------------- ## ## Setkey setkeyIfNot(DT, kCols_UD, superset.ok=TRUE) # about 11 Minutes on 232.M rows ## "Calculating first/last streams & productDELTA_byUD took 4 hours and 31 minutes" s.t({ ## Initialize the columns, especially important for the POSIX columns DT[ , `:=`( isFirstStream_byUD = FALSE # as.logical(NA) , isLastStream_byUD = FALSE # as.logical(NA) , productDELTA_byUD = as.integer(NA) , firstStream_byUD = as.POSIXct(NA) , lastStream_byUD = as.POSIXct(NA) ) ] ## Ignoring PRODUCT *AND* MOBILE, we will find the first/last stream of the MONTH (for each USER/COUNTRY) DT[ , `:=`( ## use which.min/max to find the respective extremes, f & l. ## (No risk of ties, since these are timestamped streams by a single user, and a user can only stream one at a time) ## Then, for isFirst/last, in each group, this is FALSE except for that specific index isFirstStream_byUD = {f <- which.min(firstStream_byUCnDPMb); ret <- rep(FALSE, .N); ret[f] <- TRUE; ret} , isLastStream_byUD = {l <- which.max( lastStream_byUCnDPMb); ret <- rep(FALSE, .N); ret[l] <- TRUE; ret} ## Diff of Last and First stream of a product are respectively indexed by said f/l , productDELTA_byUD = product_integer[l] - product_integer[f] ## The time stamp of the first/last stream is simply the vector of which we were finding the min/max of, indexed at f/l. , firstStream_byUD = firstStream_byUCnDPMb[f] , lastStream_byUD = lastStream_byUCnDPMb[l] ) ## Grouping by USER-COUNTRY-DATE(month) , by=kCols_UD ] ## --- I believe this is no longer needed, since I am now initializing the column beforehand --- ## ## The indexed columns will come back as an integer, so need to Convert to POSIX # DT[ , `:=`( firstStream_byUD = as.POSIXct(firstStream_byUD, origin=.origin.utc) # , lastStream_byUD = as.POSIXct(lastStream_byUD, origin=.origin.utc) # ) # ] }, msg="Calculating first/last streams & productDELTA_byUD took" ) setcolorderpt(DT, c(kCols_DCnUP, "product_integer", "mobile", "productDELTA_byUD", "firstStream_byUD", "firstStream_byUCnDPMb", "lastStream_byUCnDPMb")) notifyAndEmail("Midway 12. product deltas complete") ## SAVE ## DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb <- DT jesusForData(DT.Spotify.Counts.plusMAX_MIN_DATE.byUCnDPMb, info="End of 12. Added product delta, first and last stream flags") notifyAndEmail("End of 12. Saving complete.") # skip: # ## -------------------------------------------------------------------------------- ## # skip: # # skip: # ## Below is identical to the section above, except that this also breaks down by COUNTRY # skip: # ## Not using this for now # skip: # # skip: # ## Note that what we are counting below for first/last stream is for user-date AND **COUNTRY** # skip: # ## Thus, we cannot use this to count unique users # skip: # ## since if a user travels abroad, they will be counted once in each country (ie double counted) # skip: # ## See userid "3754da008362fd48d33e80bb3e628335" in "2013-05-01" for an example (bottom of this file) # skip: # # skip: # ## Setkey # skip: # setkeyIfNot(DT, kCols_UCnD, superset.ok=TRUE) # skip: # # skip: # s.t({ # skip: # ## Initialize the columns, especially important for the POSIX columns # skip: # DT[ , `:=`( isFirstStream_byUCnD = FALSE # as.logical(NA) # skip: # , isLastStream_byUCnD = FALSE # as.logical(NA) # skip: # , productDELTA_byUCnD = as.integer(NA) # skip: # , firstStream_byUCnD = as.POSIXct(NA) # skip: # , lastStream_byUCnD = as.POSIXct(NA) # skip: # ) # skip: # ] # skip: # # skip: # ## Ignoring PRODUCT *AND* MOBILE, we will find the first/last stream of the MONTH (for each USER/COUNTRY) # skip: # DT[ , # skip: # `:=`( # skip: # ## use which.min/max to find the respective extremes, f & l. # skip: # ## (No risk of ties, since these are timestamped streams by a single user, and a user can only stream one at a time) # skip: # # skip: # ## Then, for isFirst/last, in each group, this is FALSE except for that specific index # skip: # isFirstStream_byUCnD = {f <- which.min(firstStream_byUCnDPMb); ret <- rep(FALSE, .N); ret[f] <- TRUE; ret} # skip: # , isLastStream_byUCnD = {l <- which.max( lastStream_byUCnDPMb); ret <- rep(FALSE, .N); ret[l] <- TRUE; ret} # skip: # # skip: # ## Diff of Last and First stream of a product are respectively indexed by said f/l # skip: # , productDELTA_byUCnD = product_integer[l] - product_integer[f] # skip: # # skip: # ## The time stamp of the first/last stream is simply the vector of which we were finding the min/max of, indexed at f/l. # skip: # , firstStream_byUCnD = firstStream_byUCnDPMb[f] # skip: # , lastStream_byUCnD = lastStream_byUCnDPMb[l] # skip: # ) # skip: # ## Grouping by USER-COUNTRY-DATE(month) # skip: # , by=kCols_UCnD # skip: # ] # skip: # # skip: # ## --- I believe this is no longer needed, since I am now initializing the column beforehand --- ## # skip: # ## The indexed columns will come back as an integer, so need to Convert to POSIX # skip: # # DT[ , `:=`( firstStream_byUCnD = as.POSIXct(firstStream_byUCnD, origin=.origin.utc) # skip: # # , lastStream_byUCnD = as.POSIXct(lastStream_byUCnD, origin=.origin.utc) # skip: # # ) # skip: # # ] # skip: # }, msg="Calculating first/last streams & productDELTA_byUCnD took" ) # skip: # # skip: # ## -------------------------------------------------------------------------------- ## ### Below is an example of someone travelling aborad temporarily in one month # # Revo> DT[.("3754da008362fd48d33e80bb3e628335", c(unique(country)), as.Date("2013-05-01")), cUsing, with=FALSE, nomatch=0L] [order(firstStream_byUCnD)] # # userid month country product mobile isFirstOccurOfUser_perD isFirstStream_byUCnD firstStream_byUCnD firstStream_byUCnDP firstStream_byUCnDPMb isLastOccurOfUser_perD isLastStream_byUCnD lastStream_byUCnD lastStream_byUCnDP lastStream_byUCnDPMb # 1: 3754da008362fd48d33e80bb3e628335 2013-05-01 SE Open FALSE FALSE FALSE 2013-05-01 08:03:05 2013-05-17 20:50:46 2013-05-17 20:50:46 FALSE FALSE 2013-05-31 05:29:07 2013-05-17 23:47:16 2013-05-17 23:47:16 # 2: 3754da008362fd48d33e80bb3e628335 2013-05-01 SE Premium FALSE TRUE TRUE 2013-05-01 08:03:05 2013-05-01 08:03:05 2013-05-01 08:03:05 FALSE FALSE 2013-05-31 05:29:07 2013-05-31 05:29:07 2013-05-30 17:26:00 # 3: 3754da008362fd48d33e80bb3e628335 2013-05-01 SE Premium TRUE FALSE FALSE 2013-05-01 08:03:05 2013-05-01 08:03:05 2013-05-20 15:01:13 TRUE TRUE 2013-05-31 05:29:07 2013-05-31 05:29:07 2013-05-31 05:29:07 # 4: 3754da008362fd48d33e80bb3e628335 2013-05-01 FI Open FALSE FALSE TRUE 2013-05-16 07:58:22 2013-05-16 07:58:22 2013-05-16 07:58:22 FALSE TRUE 2013-05-16 12:37:43 2013-05-16 12:37:43 2013-05-16 12:37:43