# ~/git/orch/src/Discover_Weekly/create user_counts tables programatically.r ### This file iterates over a set of column groups, then creates tables ## Where each table counts the total number of users and sterams per day, by group ## as well as cumulative sum at any given day if (FALSE) { ## screen -xRR UserCounts; R setGitBranchToSystem(); .g() source ("~/git/orch/src/Discover_Weekly/create user_counts tables programatically.r") } setScience("Discover_Weekly", subProj="UserCounts") setGitBranchToSystem(); .g() wh <- "SpotifyAggregates" dbname <- "prod" sinkFile <- sinkOn()$filename setSnowflake(wh=wh, dbname=dbname, start=TRUE) sfWaitForWarehouse(wh=wh, N.seconds=20, max_iterations=500, verbose=TRUE) verbose.qry <- !interactive() verbose.qry <- TRUE TEST <- FALSE ## FULL DETAULS schema <- "spot_user_counts" tbl.detailed <- "user_minmax_timestamps_detailed" schema.raw <- "spotify" tbl.raw <- "sos_from_raw_view" if (TEST) tbl.detailed <- "user_minmax_timestamps_detailed_TEST" ### ---- FIRST CREATE DETAILED COUNTS ---------- ## ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## UNIQUE USERS, TRACK, SOURCE, SOURCE_URI, TIMESTAMP, etc... ## spot_user_counts.user_minmax_timestamps_detailed ## If the detailed table does not exist, or it is more than 3 days old, (re)create it. if (!qTableExists(tbl=tbl.detailed, schema=schema, dbname=dbname, wh=wh) || as.Date(qMaxDate(tbl=tbl.detailed, schema=schema, dbname=dbname, wh=wh, dateCols="last_updated", verbose=FALSE)) < (today() - 3) ) { message("Detailed Table does not exist -- will create it") colsToPull <- c("album_code", "isrc", "track_album_artist", "country_code", "anonymized_uid", "user_access", "user_type", "user_birthyear", "source", "playlist_is_unknown", "playlist_owner_group", "playlist_owner_id_from_source_uri", "playlist_uri", "playlist_id", "os_group", last_updated="current_timestamp()") colsWithaggFunc <- c(first_tmstamp="min(activity_date)", last_tmstamp="max(activity_date)") colsToAgg <- c(streams="*") aggFunc <- "count" orderby <- NULL ## Slows down unnecessarilly if (!TEST) c("album_code", "isrc", "anonymized_uid", "source","last_updated") limit <- NULL ## if ( TEST) 25000 maxDate <- if ( TEST) as.Date("2014-12-29") dateCol <- "activity_date" if (TEST) { where <- list("(user_access != 'deleted') AND (1=0 ", isrc = c("DKEFA0101601", "USCPT1110003", "USA6B0300201"), anonymized_uid=c("714ed99300c4ae75158c296981dc7871", "89684fbd96917f4acad92e44bbd5377d", "06260cb42f0d2be72f495416b0c970ae", "0e1e5866cd1de62c68fb3dcf1c536bae") , os_group=c("iOS/OSX"), "1=0)" , conjunction="OR") } else { where <- "user_access != 'deleted'" } qry.detailed <- makeQry(tbl=tbl.raw, schema=schema.raw, colsToPull=colsToPull, colsToAgg=colsToAgg, colsWithaggFunc=colsWithaggFunc, aggFunc=aggFunc, limit=limit, orderby=orderby, maxDate=maxDate, dateCol=dateCol, whereIn=where, lte.maxDate="<") sfPopulateTable(qry=qry.detailed, tbl=tbl.detailed, schema=schema, over=TRUE) # qry.detailed <- # "SELECT # album_code, # isrc, # anonymized_uid, # user_access, # user_type, # user_birthyear, # source, # playlist_is_unknown, # playlist_owner_group, # playlist_owner_id_from_source_uri, # playlist_uri, # playlist_id, # os_group, # country_code, # min(tmstamp) AS first_tmstamp, # max(tmstamp) AS last_tmstamp, # count(*) AS streams, # current_timestamp() AS last_updated # FROM spotify.sos_from_raw_view # WHERE user_access != 'deleted' # GROUP BY 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, last_updated # ORDER BY 1, 2, 3, 7, 11" # # sfPopulateTable(qry=qry.detailed, tbl=tbl.detailed, schema=schema, over=TRUE) message("Done creating the detailed table. Will now confirm stream counts with the original raw data") ## Pull all of the tbl.raw figures first, since Snowflake probably cached most of this streams_actual <- makeQry(tbl=tbl.raw, schema=schema.raw, whereIn=where, colsToAgg="*", aggFunc="count", limit=NULL) %>% sfQry(verbose=FALSE) %>% setkey() ## These are for debugging / troubleshooting streams_actual.by_source <- makeQry(tbl=tbl.raw, schema=schema.raw, colsToPull=c("source"), whereIn=where, colsToAgg="*", aggFunc="count", limit=NULL, orderby="colsToPull") %>% sfQry(verbose=FALSE) %>% setkey() streams_actual.by_source_os <- makeQry(tbl=tbl.raw, schema=schema.raw, colsToPull=c("source", "os_group"), whereIn=where, colsToAgg="*", aggFunc="count", limit=NULL, orderby="colsToPull") %>% sfQry(verbose=FALSE) %>% setkey() ## Then pull the "streams_in_new_tbl" counts, one part at a time, after confirming streams_in_new_tbl <- makeQry(tbl=tbl.detailed, schema=schema, colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% setkey() stopifnot(streams_in_new_tbl == streams_actual) message("Sterams, overall confirmed!!") streams_in_new_tbl.by_source <- makeQry(tbl=tbl.detailed, schema=schema, colsToPull=c("source"), colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% setkey() # setkey(streams_in_new_tbl.by_source) # setkey(streams_actual.by_source) stopifnot(streams_in_new_tbl.by_source == streams_actual.by_source) message("Sterams by source confirmed!!") streams_in_new_tbl.by_source_os <- makeQry(tbl=tbl.detailed, schema=schema, colsToPull=c("source", "os_group"), colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% setkey() # setkey(streams_in_new_tbl.by_source_os) # setkey(streams_actual.by_source_os) stopifnot(streams_in_new_tbl.by_source_os == streams_actual.by_source_os) message("Sterams by source + os_group confirmed!!") } else { verboseMsg(TRUE, "No need to recreate tbl.detailed ", tbl.detailed) } ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## uid_col <- "anonymized_uid" colGroups <- list( upc = "album_code" , isrc = "main_isrc" , freepaid = "user_type" , country = "country_code" , os = "os_group" , source = "source" , playlist = c("playlist_id", "playlist_uri", "playlist_owner_id_from_source_uri", "playlist_is_unknown") ## TRACK_ALBUM_ARTIST can change, for the same ISRC, from one date to another # , artist = "track_album_artist" ) groups <- lapply(seq(length(colGroups)), function(m) combinat::combn(x=names(colGroups), m=m, simplify=FALSE)) %>% unlist(recursive=FALSE) ## No need for playlist without source, since having the two together will not change the counts for playlists groups %<>% lapply(function(x) if ("playlist" %in% x && "source" %ni% x) NULL else x) %>% removeNullsAndBlanksFromList ## No need for upc/isrc with artist, since this will just split up a upc/isrc anytime metadata has changed groups %<>% lapply(function(x) if ("artist" %in% x && any(c("upc", "isrc") %in% x)) NULL else x) %>% removeNullsAndBlanksFromList ## artist is pretty sloppy. Therefore, only include a few specific groups, no need for all combinations groups %<>% c(., list( c("artist", "freepaid") , c("artist", "freepaid", "country_code") , c("artist", "country_code") , c("artist", "country_code", "source", "playlist") , c("artist", "source", "playlist") )) ## add uid_col to each group. Add one stand-alone group, for "daily_totals" groups %<>% c(list(), .) %>% lapply( function(x) c("uid", x)) %>% c("uid", .) ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## KEEPING IT A SHORTER LIST FOR NOW groups <- list( c("isrc") , c("isrc", "playlist", "source") , c("isrc", "source") , c("isrc", "country") , c("isrc", "country", "source", "playlist") ) ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## calculate the streams_expected, now once, so to not have to run this each time. if (exists("streams_in_new_tbl")) { streams_expected <- unlist(streams_in_new_tbl) } else { streams_expected <- makeQry(tbl=tbl.detailed, schema=schema, colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist() } ## Picking out the most important ones main_groups <- list(c("uid", "isrc"), c("uid", "country"), c("uid", "isrc", "source", "playlist")) length(groups %>% setdiff(main_groups)) ## for testing ## OS_GROUP # grp <- groups[[6]] ## USER COUNTS for (grp in groups) { catn("USER COUNTS: ", commaSep(grp)) tmp_colsToPull <- c(uid=uid_col, colGroups)[grp] %>% unlist(use.names=FALSE) if ("album_code" %in% tmp_colsToPull) { selfname_(tmp_colsToPull) names(tmp_colsToPull)[tmp_colsToPull == "album_code"] <- "upc" } try(create_spotify_user_counts_table(colsToPull=tmp_colsToPull, tbl_in=tbl.detailed, schema=schema, confirm_streams=TRUE, streams_expected=streams_expected, having_clause=NULL, minDate=NULL, verbose.qry=verbose.qry)) } for (grp in sample(groups)) { catn("STREAM COUNTS: ", commaSep(setdiff(grp, "uid"))) tmp_colsToPull <- c(uid=uid_col, colGroups)[grp] %>% unlist(use.names=FALSE) if ("album_code" %in% tmp_colsToPull) { selfname_(tmp_colsToPull) names(tmp_colsToPull)[tmp_colsToPull == "album_code"] <- "upc" } try(create_spotify_stream_counts_table(colsToPull=tmp_colsToPull, confirm_streams=TRUE, schema_in="spotify", having_clause=NULL, minDate=NULL, verbose.qry=verbose.qry)) } ### DEBUG if (FALSE) { DT.check <- sfQry("SELECT * FROM prod.spot_user_counts.by_upc_isrc_source_playlist WHERE ISRC='DED831500528' ") debugOn("user_counts") where__for_debugging <- list(ISRC='DED831500528') create_spotify_user_counts_table(colsToPull=tmp_colsToPull, tbl_in=tbl.detailed, schema=schema, confirm_streams=TRUE, streams_expected=streams_expected, having_clause=NULL, minDate=NULL, verbose.qry=verbose.qry, where__for_debugging=where__for_debugging) catn(verboseQry(qry.cumsum)) debugOff("user_counts") } # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # verboseMsg(TRUE, "DONE RUNNING 'create user_counts tables programatically.r'", func="message") sinkOff() showSink()