# ~/git/orch/src/Discover_Weekly/supportFns/create_spotify_user_counts_table supportFns.r create_spotify_user_counts_table <- function( colsToPull , colsWithaggFunc = c(date_first_streamed = "friday_weekof(min(first_tmstamp::date))") ## , date_last_streamed = "friday_weekof(max(last_tmstamp::date))" , colsToAgg = c("streams") , tbl_out = "..auto.." , aggFunc = "sum" , tbl_in = "user_minmax_timestamps_detailed_test" , schema = "spot_user_counts" , schema_in = schema , confirm_streams = TRUE , uid_columns=c("anonymized_uid", "user_id", "uid") , minDate = "friday_weekof('2015-01-01'::date)" , having_clause = if (length(minDate) == 1) sprintf("date_first_streamed > %s", ifelse(grepl("^\\d", minDate), sprintf("'%s'", minDate), minDate)) , streams_expected = makeQry(tbl=tbl_in, schema=schema_in, colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist() , where__for_debugging=NULL , verbose.qry=FALSE ) { ## ============================================================================================================== ## ## ----------------------------------------------------------------------------------------------------------- # ## This requires four queries ## ----------------------------------------------------------------------------------------------------------- ## WITH CTE ( ## qry.counts ## FROM ( ## qry.inner ## ) ## ) ## ## qry.cumsum ## FROM ( ## qry.self_join ## FROM CTE ## ) ## ## ## Recaping: ## ----------------------------------------------------------------------------------------------------------- ## 1. inner ## 2. counts ## 3. self_join ## 4. cumsum ## ## Details: ## ----------------------------------------------------------------------------------------------------------- ## 1. inner :: counts the number of streams per user per group; In this table, every row in a group is a unique user ## 2. counts :: by removing user, we can count rows to get a user-count, and sum streams to get total streams, per group ## Tables 3 & 4 could be a single table if we were able to do a cumsum. But since not available, we have to do a self-join. ## We cant GROUP BY in the self-join query, since that will remove the 'duplicate' rows which we in fact want. ## ie, it will yield only one row of T2.streams per each date+group+streams combination of T1. ## What we want, for the cumsum, is a series of T2 streams per each of above, where the series has all of the dates prior to T1.date ## See Example below ## 3. self_join :: A blow-out of table 2, with the same columns plus a second column of each, streams and new_users ## We filter to only those rows where T1.date >= T2.date (meaning those rows where T2.date is in the future ## realitve to T1 are not considered. T2 contributes the columns that will be summed for cumsum) ## 4. cumsum :: Now we simply aggregate the cumsum columns, grouping by all the group columns AND the T1 streams and user counts ## Here, we also add some additional metrics, such as lag, diff, perc_increase, accel score, scaled accel_score ## ## ## Example of 3, self_join: ## ----------------------------------------------------------------------------------------------------------- ## source os_group T1.date T2.date T1.new_users T1.streams T2.streams T2.new_users ## 1: album iOS/OSX 2014-08-25 2014-08-25 1467767 217837384 217837384 1467767 -- start of cumsum() ## 2: album iOS/OSX 2014-09-01 2014-08-25 1061783 93937594 217837384 1467767 -- start of cumsum() ## 3: album iOS/OSX 2014-09-01 2014-09-01 1061783 93937594 93937594 1061783 ## 4: album iOS/OSX 2014-09-08 2014-08-25 935913 67045056 217837384 1467767 -- start of cumsum() ## 5: album iOS/OSX 2014-09-08 2014-09-01 935913 67045056 93937594 1061783 ## 6: album iOS/OSX 2014-09-08 2014-09-08 935913 67045056 67045056 935913 ## 7: album iOS/OSX 2014-09-15 2014-08-25 853672 53645944 217837384 1467767 -- start of cumsum() ## 8: album iOS/OSX 2014-09-15 2014-09-01 853672 53645944 93937594 1061783 ## 9: album iOS/OSX 2014-09-15 2014-09-08 853672 53645944 67045056 935913 ## 10: album iOS/OSX 2014-09-15 2014-09-15 853672 53645944 53645944 853672 ## 11: album iOS/OSX 2014-09-22 2014-08-25 751356 42891128 217837384 1467767 -- start of cumsum() ## 12: album iOS/OSX 2014-09-22 2014-09-01 751356 42891128 93937594 1061783 ## 13: album iOS/OSX 2014-09-22 2014-09-08 751356 42891128 67045056 935913 ## 14: album iOS/OSX 2014-09-22 2014-09-15 751356 42891128 53645944 853672 ## 15: album iOS/OSX 2014-09-22 2014-09-22 751356 42891128 42891128 751356 ## 16: album iOS/OSX 2014-09-29 2014-08-25 688565 36583534 217837384 1467767 -- start of cumsum() ## 17: album iOS/OSX 2014-09-29 2014-09-01 688565 36583534 93937594 1061783 ## 18: album iOS/OSX 2014-09-29 2014-09-08 688565 36583534 67045056 935913 ## 19: album iOS/OSX 2014-09-29 2014-09-15 688565 36583534 53645944 853672 ## 20: album iOS/OSX 2014-09-29 2014-09-22 688565 36583534 42891128 751356 ## 21: album iOS/OSX 2014-09-29 2014-09-29 688565 36583534 36583534 688565 ## 22: artist iOS/OSX 2014-08-25 2014-08-25 1728892 244213684 244213684 1728892 -- start of cumsum() ## 23: artist iOS/OSX 2014-09-01 2014-08-25 1161184 99781562 244213684 1728892 -- start of cumsum() ## 24: artist iOS/OSX 2014-09-01 2014-09-01 1161184 99781562 99781562 1161184 ## 25: artist iOS/OSX 2014-09-08 2014-08-25 971502 67948373 244213684 1728892 -- start of cumsum() ## 26: artist iOS/OSX 2014-09-08 2014-09-01 971502 67948373 99781562 1161184 ## 27: artist iOS/OSX 2014-09-08 2014-09-08 971502 67948373 67948373 971502 ## 28: artist iOS/OSX 2014-09-15 2014-08-25 860937 52572977 244213684 1728892 -- start of cumsum() ## 29: artist iOS/OSX 2014-09-15 2014-09-01 860937 52572977 99781562 1161184 ## 30: artist iOS/OSX 2014-09-15 2014-09-08 860937 52572977 67948373 971502 ## 31: artist iOS/OSX 2014-09-15 2014-09-15 860937 52572977 52572977 860937 ## 32: artist iOS/OSX 2014-09-22 2014-08-25 766485 42556333 244213684 1728892 -- start of cumsum() ## 33: artist iOS/OSX 2014-09-22 2014-09-01 766485 42556333 99781562 1161184 ## 34: artist iOS/OSX 2014-09-22 2014-09-08 766485 42556333 67948373 971502 ## 35: artist iOS/OSX 2014-09-22 2014-09-15 766485 42556333 52572977 860937 ## 36: artist iOS/OSX 2014-09-22 2014-09-22 766485 42556333 42556333 766485 ## 37: artist iOS/OSX 2014-09-29 2014-08-25 683490 35753486 244213684 1728892 -- start of cumsum() ## 38: artist iOS/OSX 2014-09-29 2014-09-01 683490 35753486 99781562 1161184 ## 39: artist iOS/OSX 2014-09-29 2014-09-08 683490 35753486 67948373 971502 ## 40: artist iOS/OSX 2014-09-29 2014-09-15 683490 35753486 52572977 860937 ## 41: artist iOS/OSX 2014-09-29 2014-09-22 683490 35753486 42556333 766485 ## 42: artist iOS/OSX 2014-09-29 2014-09-29 683490 35753486 35753486 683490 ## source os_group date t2_date new_users streams t2_streams t2_new_users ## ## ============================================================================================================== ## ## ----------------------------------------------------------------------------------------------------------- ## ============================================================================================================== ## if (aggFunc != "sum") warning("Aggregation may be incorrect when using aggFunc other than 'sum'") if (!any(uid_columns %in% colsToPull)) warning("There is no USER ID column in colsToPull. Was this intentional?\n\nHINT: Not having a user_id column is going to cause all of the dates to be the first by group, regardless of user") if (!is.null(where__for_debugging)) message("REMINDER: The WHERE clause should only be used for debugging. It will apply to the inner table") if (tbl_out == "..auto..") tbl_out <- create_counts_table_name_from_colsToPull(colsToPull=colsToPull, uid_columns=uid_columns) if (grepl("_test$", tbl_in, ignore.case=TRUE) && !grepl("_test", tbl_out, ignore.case=TRUE)) { verboseMsg(TRUE, "Pasting '_test' to name of tbl_out") tbl_out %<>% paste0("_TEST") } ## ------ INNER QUERY --------------------------------- qry.inner <- makeQry(tbl=tbl_in, schema=schema_in, colsToPull=colsToPull, colsToAgg=colsToAgg, colsWithaggFunc=colsWithaggFunc, aggFunc=aggFunc, having=having_clause, where=where__for_debugging, order=NULL, limit=NULL) ## ------ COUNTS QUERY --------------------------------- ## combine the agg'd columns with colsToPull, but remove the "last streamed" column, the last_updated and, most importantly, the user colsToPull.counts <- c(colsWithaggFunc, colsToPull) %>% colNamesFromVector() %>% setdiff(., c("date_last_streamed", "last_updated", "last_tmstamp", uid_columns)) %>% tolower %>% {setNames(obj=., nm=ifelse(. == "date_first_streamed", "date", .))} # %>% c(cumulative_new_users= default_cum_user_count) ## Each row is one user colsWithaggFunc.counts <- c(new_users = "count(*)") qry.counts <- makeQry(tbl=qry.inner, schema=NULL, colsToPull=colsToPull.counts, colsToAgg=colsToAgg, colsWithaggFunc=colsWithaggFunc.counts, aggFunc=aggFunc, limit=NULL) ## No need to order order=colsToPull.counts ## The WITH ... CTE clause ## This will actually be applied in the makeQry() call for the final, ie the cumsum, query ## However, it's name will be referenced in the third, ie the self_join, query with.cumsum <- c(CTE_Counts=qry.counts) ## ------ SELF_JOIN QUERY --------------------------------- ## Will agg the original columns, plus aggregate new_users ## Preface colsToAgg with T2, and colsToPull with T1 ## The columns that will be cumsumd are "new_users", and the colsToAgg, assuming the aggFunc is itself sum ## (in the default, the colsToAgg will simply be "streams" and the aggFunc will be "sum") colsThatWillBeCumSumd <- c(if (aggFunc == "sum") colsToAgg, "new_users") %>% {setNames(obj=paste0("T2.", .), nm=paste0("T2_", .))} ## The colsToPull aka, the grouping columns in the final query, will be the colsToPull from previous, ## plus any cols we previously aggregated (generally thats "new_users" and "streams") colsToPull.self_join <- c(colsToPull.counts, colsWithaggFunc.counts, colsToAgg) %>% colNamesFromVector # %>% paste0("T1.", .) ## The main part of the "ON" clause is the T1.date >= T2.date, which appears below ## However, in addtion to this, we must also join on each group ## Since NULL values are possible in any group, we need to account for them with equal_null(X, Y) ## ----- ## HOWEVER, due to the snowflake bug in equal_null(), we have to get creative and use NVL() ## which requires us to know the data type of each column ## --------- ## The columns to join on are the group columns (from counts query), less the date column colsToJoinOn <- colsToPull.counts %>% colNamesFromVector %>% setdiff("date") ## Create the additional ON_AND clause; If there are no such columns, it should be blank (hence ifelse()) ON_AND.self_join <- ifelse (!length(colsToJoinOn), "", { # colsToPull.self_join %>% setdiff("date") %>% sprintf("equal_null(T1.%s, T2.%1$s)", .) %>% paste(" AND ", .) %>% pasteC ## DOESN'T WORK. BUGGY ## ------------------------------------------------- ## frmt.null_joiner <- "equal_null(T1.%s, T2.%1$s)" ## equal_null() is buggy, thus using NVL() colsToJoinOn %>% sprintf(frmt.null_joiner, .) %>% paste("\n AND ", .) %>% pasteC ## ------------------------------------------------- ## ## TEMP FIX ## ------------------------------------------------- ## frmt.null_joiner.numeric <- "NVL(T1.%s, -123.1230321) = NVL(T2.%1$s, -123.1230321)" frmt.null_joiner.string <- "NVL(T1.%s, 'dummy_dummy_value___xyz123') = NVL(T2.%1$s, 'dummy_dummy_value___xyz123')" ## Which columns to use the numeric version. Technically UPC is numeric and album_code is string; but should only use album_code anyway tmp.known_numeric_cols <- c("playlist_is_unknown", "user_birthyear", "length_seconds", "row_id") colsToJoinOn %>% {sprintf(ifelse(. %in% tmp.known_numeric_cols, frmt.null_joiner.numeric, frmt.null_joiner.string), .)} %>% paste("\n AND ", .) %>% pasteC() ## ------------------------------------------------- ## }) ## Cobine the three pieces of the join statement ## The acutal join clause; the date filter; the ON_AND for the rest of the group columns join_statement.self_join <- sprintf("\n INNER JOIN %s T2\n ON T1.date >= T2.date %s", names(with.cumsum), ON_AND.self_join) ## The table is the name of the CTE from the WITH clause tbl.self_join <- c(T1=names(with.cumsum)) qry.self_join <- makeQry(with=NULL, tbl=tbl.self_join, colsToPull=colsToPull.self_join %>% paste0("T1.", .) %>% c(colsThatWillBeCumSumd), colsToAgg=NULL, join=join_statement.self_join, aggFunc=NULL, prependCols.with.tbl=FALSE, nogroupby=TRUE, limit=NULL) ## ------ CUMULATIVE SUM QUERY --------------------------------- ## The cumsum query is pulling the same columns as the self_join query, less the T2 columns, which will instead be aggregated colsToPull.cumsum <- colsToPull.self_join ## Grouping by the same cols. We want to be explicit here. We might be adding in the addl_cols, which makeQry() will otherwise add these to the Group By clause and there might be a loss of efficiency groupby.cumsum <- colsToPull.self_join ## Agg the colsThatWillBeCumSumd colsToAgg.cumsum <- colsThatWillBeCumSumd %>% colNamesFromVector() %>% {setNames(., nm=gsub("T2", "cumulative", .))} ## CREATE AN ACCELERATION SCORE ## - For now, Only do so if the only colsToAgg is "streams" ## This could be expanded to other colsToAgg if needed, so long as the aggFunc is "SUM" if (isTRUE(all("streams" == colsToAgg))) { cols <- c("streams", "new_users") make_accel_cols <- function(col, dateCol="date", ctp) { # ctp %<>% setdiff(dateCol) %>% setdiff(cols) partition_clause <- ctp %>% setdiff(dateCol) %>% setdiff(cols) %>% {ifelse(length(.), sprintf("PARTITION BY %s ", commaSep(.)), "")} LG <- sprintf("(LAG(%s, 1, NULL) OVER (%sORDER BY %s ASC))", col, partition_clause, dateCol) diffCol <- sprintf("(%2$s - %1$s)", LG, col) %>% setNames(obj=., nm=paste0(col, "_increase")) percCol <- sprintf("(%2$s - %1$s)/%1$s", LG, col) %>% setNames(obj=., nm=paste0(col, "_perc_increase")) accelCol <- sprintf("score_accel_before_after(%1$s, %2$s)", LG, col) %>% setNames(obj=., nm=paste0(col, "_accel_score")) scaledAccelCol <- sprintf("scaled_score_accel_before_after(%1$s, %2$s)", LG, col) %>% setNames(obj=., nm=paste0(col, "_accel_score_scaled")) return(c(diffCol, percCol, accelCol, scaledAccelCol)) } addl_cols <- lapply(unname(cols), make_accel_cols, ctp=colsToPull.cumsum) %>% unlist(recursive=FALSE, use.names=TRUE) } else { addl_cols <- NULL } qry.cumsum <- makeQry(with=with.cumsum, tbl=qry.self_join, colsToPull=c(colsToPull.cumsum, addl_cols), colsToAgg=colsToAgg.cumsum, aggFunc="sum", prependCols.with.tbl=FALSE, groupby=groupby.cumsum, limit=NULL) # if (verbose.qry) # catn(verboseQry(qry.cumsum, max.width=150, max.lines=55, add.dots=FALSE)) browser(expr=inDebugMode(c("user_counts", "create_spotify_user_counts_table")), text="in create_spotify_user_counts_table() right before calling sfPopulateTable()") ## catn(verboseQry(qry.cumsum)) sfPopulateTable(tbl=tbl_out, schema=schema, qry=qry.cumsum, over=TRUE, verbose.rows=TRUE, verbose=verbose.qry) if (confirm_streams) { tryCatch({ streams_users <- makeQry(tbl=tbl_out, schema=schema, colsToAgg=c("streams", "new_users"), aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist() streams <- streams_users[["streams"]] new_users <- streams_users[["new_users"]] msg <- sprintf("Total number of streams %s (%s). Total New Users is %s", ifelse(streams == streams_expected, "confirmed", "is *WRONG*"), formnumb(streams, round=FALSE), formnumb(new_users, round=-3)) verboseMsg(TRUE, msg, func="message", minw=100) }, error=function(e) { warning("Was unable to confirm streams due to an error. The error was caught and execution unaffected. Caught error message: \n\t", e$message, "\n", call.=FALSE); }) } return(tbl_out) } ## EXAMPLE ## # ---------------------------------------------------------- # # WITH unnamed_cte AS ( # SELECT date_first_streamed AS date, isrc, source, playlist_is_unknown, playlist_uri # , SUM(streams) AS streams # , COUNT(*) AS new_users, seq8() AS row_number # FROM # ( # SELECT isrc, anonymized_uid, source, playlist_is_unknown, playlist_uri # , SUM(streams) AS streams # , MIN(first_tmstamp::date) AS date_first_streamed # , MAX(last_tmstamp::date) AS date_last_streamed # FROM spot_user_counts.user_minmax_timestamps_detailed_test # -- DEBUGGING # WHERE isrc = 'TR0061500165' AND first_tmstamp < '2015-08-30' # GROUP BY 1, 2, 3, 4, 5 # ) # GROUP BY 1, 2, 3, 4, 5 # ORDER BY date,isrc,source,playlist_is_unknown,playlist_uri # ) # # SELECT # T1.date # --------- # , T1.isrc # , T1.source # , T1.playlist_is_unknown # , T1.playlist_uri # --------- # , T1.streams # , T1.new_users # , sum(T2.new_users) AS cusers # , sum(T2.streams) AS cstreams # FROM unnamed_cte T1 # FULL JOIN unnamed_cte T2 # ON T1.date >= T2.date # AND equal_null(T1.isrc, T2.isrc) # AND equal_null(T1.source, T2.source) # AND equal_null(T1.playlist_is_unknown, T2.playlist_is_unknown) # AND equal_null(T1.playlist_uri, T2.playlist_uri) # WHERE T1.isrc = 'TR0061500165' AND T1.date < '2015-08-30' # GROUP BY T1.date, T1.isrc, T1.source, T1.playlist_is_unknown, T1.playlist_uri, T1.streams, T1.new_users # order by source asc, 1, 2, 3 # ; # ----------------------------------------------------------