setScience("Spotify_Accounting_ETL", subl=FALSE) library(bit64) library(reshape2) setDBall(cluster=4) labels_using <- c(8869,10693,17048,8099,8673,15063,20325,19155,18805,6971,21384,18080,21903) countries_using <- c("US", "DE", "SE", "FR", "MX", "AU", "ES", "DK", "FI", "NL", "NO") countries_using <- c("US", "SE", "GB", "FR", "DE") countries_using <- c("US", "DE", "SE", "FR", "AU", "ES", "DK", "FI", "NL", "NO") countries_using <- c("US", "DE", "SE", "FR", "AU", "DK", "FI", "NL", "NO") minDate <- "2013-04-01" frmt.create_with_label <- { " CREATE TABLE bi.%s ( owner VARCHAR (11), isdistributor VARCHAR (2), labelid INT4, genreid INT2, upc NUMERIC (22, 6), customerid VARCHAR(35), product VARCHAR(2), country VARCHAR(3), streams BIGINT ) DISTSTYLE EVEN SORTKEY ( upc ) "} tbl_spot <- "staging_raw_spotify" maxDates <- seq.Date(as.Date("2013-09-01"), by="3 months", length.out=12) - 1 maxDates <- maxDates[maxDates < today()] nms <- format(maxDates, "upto_%Y%m%d") setattr(maxDates, "names", nms) setattr(nms, "names", nms) Qrys.full <- lapply(nms, function(nm) { setQry(sprintf( "SELECT L.owner, L.isdistributor, TT.* FROM ( SELECT R.labelid, R.genreid, S.* FROM ( %s ) S LEFT JOIN production.dim_release R ON S.upc = R.releaseid ) TT LEFT JOIN production.dim_label L ON TT.labelid = L.labelid ", makeQry(tbl=tbl_spot, schema="production", colsToPull=c("upc", "customerid" , product="CASE WHEN product IN ('O', 'F') THEN 'F' ELSE PRODUCT END", "country"), colsToAgg=c(streams="*"), aggFunc="count", distinct=FALSE, limit=NULL, maxDate=maxDates[[nm]], dateCol="download_date", minDate=minDate) )) }) setDBall(cluster=4) for (nm in names(maxDates)) { bucket <- s3_p("spotify", "20150413_STREAMS_per_users_per_release_country", nm, dateStamp("processed")) unloadQry(qry=Qrys.full[[nm]], bucket=bucket) } setDBall(cluster=9) for (nm in names(maxDates)) { tbl <- paste0("streams_by_labels_and_customers_", nm) cat("tbl = ", tbl, "\n") try(runQry(sprintf("DROP TABLE bi.%s", tbl))) runQry(sprintf(frmt.create_with_label, tbl)) bucket <- s3_p("spotify", "20150413_STREAMS_per_users_per_release_country", nm) loadFromBucket(bucket=bucket, tbl=tbl, schema="bi") } ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## PULL THE DATA AND RBIND IT kCols <- c("owner", "labelid", "label_name", "product", "country", "upto") ll_DT.stream_customer_counts_by_label <- emptylist(maxDates) for (nm in names(maxDates)) { tbl <- paste0("streams_by_labels_and_customers_", nm) Qry <- makeQry(tbl=tbl, schema="bi", colsToPull=c("owner", "labelid", "product", "country", "upto"=format(maxDates[[nm]], "'%Y-%m-%d'") ) , colsToAgg=setNames( nm = c("unique_users", "total_streams") , obj = c("COUNT(DISTINCT customerid)", "streams")) , aggFunc="sum", limit=NULL, key="colsToPull") Qry <- gsub("SUM\\(COUNT\\(DISTINCT customerid\\)\\)", "COUNT(DISTINCT customerid)", Qry) ll_DT.stream_customer_counts_by_label[[nm]] <- runQry(Qry, verbose=TRUE) } &&&&&&& HARDCODED - WHY? stop("REMINDER TO RICK 2015-06-16 -- DATES ARE HARDCODED -- WHAT DO YOU WANT? ") runQry(paste0("SELECT * FROM bi.", tbl, " WHERE customerid = '54b05482c7465025128ad6887040763e' LIMIT 25")) runQry(paste0("SELECT * FROM bi.", paste0("streams_by_labels_and_customers_", "upto_20140331 WHERE customerid = '54b05482c7465025128ad6887040763e'"), " LIMIT 25")) ## Flatten: DT.stream_customer_counts_by_label <- rbindlist(ll_DT.stream_customer_counts_by_label) ## CLEANUP DT.stream_customer_counts_by_label[, upto := as.Date(upto)] DT.stream_customer_counts_by_label[product == "F", product := "A"] addLabelName.labelid_(DT.stream_customer_counts_by_label) ## Key setkeyIfNot(DT.stream_customer_counts_by_label, kCols, verbose=FALSE, organize=TRUE) ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## # DT.stream_customer_Spotify <- Reduce(f=function(L, R) data.table:::merge.data.table(L, R, all=TRUE), DT.stream_customer_Spotify <- rbindlist(lapply(rev(names(maxDates)), function(nm) { # cols <- paste0(c("unique_users_", "total_streams_"), nm) cols <- c("unique_users", "total_streams") DT.revshare[month >= minDate & month <= maxDates[[nm]] & product %in% c("A", "P", "U") , c(list(upto=maxDates[[nm]]), setNames(nm=cols, obj=lapply(.SD, function(x) sumn(as.numeric(x))))) , keyby=list(product, country=country_code) , .SD=c("active_users", "total_tracks")] })) DT.stream_customer_Spotify[, `:=`(owner = "Spotify", labelid=-1, label_name = "Spotify Overall")] setkeyIfNot(DT.stream_customer_Spotify, kCols, verbose=FALSE, organize=TRUE) jesusForData("DT.stream_customer_counts_by_label") jesusForData("DT.stream_customer_Spotify") ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## --------------------------------------------------------------- ## ## --- CLEANING ------------------------------------------------ ## ## --------------------------------------------------------------- ## # old # reshapeCounts <- function(DT) { # old # # old # if (!haskey(DT)) # old # stop ("DT has no key set") # old # # old # DT.molten <- melt(DT, id.vars=key(DT)) # old # DT.molten[, c("var", "upto") := as.data.table(do.call(rbind, strsplit(as.character(variable), "_upto_")))] # old # DT.molten[, upto := as.Date(upto, format="%Y%m%d")] # old # DT.molten[, variable := NULL] # old # newKey <- c(key(DT), "upto") # old # # old # ret <- dcast.data.table(DT.molten, makeFormula(newKey, "var")) # old # setkeyv(ret, newKey) # old # # old # return(ret) # old # } rbind_WW_total <- function(DT) { key.bak <- key(DT) if ("country" %ni% names(DT)) stop ("country is not in names(DT)") if (is.null(key.bak)) stop ("The key for DT is not yet set") tmp_DT.totals <- DT[, lapply(.SD, sumn), by=setdiff(key.bak, "country"), .SDcols=setdiff(names(DT), c("country", key.bak))] [, country := "WW"] matchKey(DT, tmp_DT.totals, keyCols=key.bak, organize=TRUE, verbose=FALSE) ret <- rbind(DT, tmp_DT.totals) setkeyv(ret, key.bak) } calculateSPU_ <- function(DT) { if (is.null(key(DT)) || tail(key(DT), 1) != "upto") stop ("The key of DT does NOT end with 'upto'") DT[, streams_per_user := total_streams / unique_users] DT[, quarterly_change_in_spu := percentIncrease(streams_per_user), by=setdiff(key(DT), "upto")] } calculateRank_ <- function(DT) { ## shortcircuit for Spotify if (DT[1, owner] == "Spotify" && all(DT$owner == "Spotify")) DT[, `:=`(rank_addsupp = 0, rank_premium = 0)] else { key.bak <- key(DT) tmp_DT.label_rank <- DT[, list(label_2014_fall_Streams = sumn(total_streams[upto == "2014-09-30"]) - sumn(total_streams[upto == "2014-06-30"])), keyby=list(owner, labelid, product)] tmp_DT.label_rank[product == "A", rank_addsupp := rank(-label_2014_fall_Streams)] tmp_DT.label_rank[product == "P", rank_premium := rank(-label_2014_fall_Streams)] tmp_DT.label_rank <- tmp_DT.label_rank[, lapply(.SD, removeNA), keyby=list(owner, labelid), .SDcols=c("rank_addsupp", "rank_premium")] ## Add the rank columns matchKey(DT, tmp_DT.label_rank, keyCols=c("owner", "labelid"), superset.ok=TRUE, verbose=FALSE) DT[tmp_DT.label_rank, c("rank_addsupp", "rank_premium") := list(rank_addsupp, rank_premium)] rm(tmp_DT.label_rank) setkeyIfNot(DT, key.bak, verbose=FALSE) } return(invisible(DT)) } DT.Spotify_molten <- rbind_WW_total(DT.stream_customer_Spotify) DT.Orchard_molten <- rbind_WW_total(DT.stream_customer_counts_by_label) calculateRank_(DT.Spotify_molten) calculateRank_(DT.Orchard_molten) ## Add SPU info calculateSPU_(DT.Spotify_molten) calculateSPU_(DT.Orchard_molten) ## Create Molten data DT.Molten <- rbind(DT.Spotify_molten, DT.Orchard_molten, use.names=TRUE) setkeyIfNot(DT.Molten, kCols, verbose=FALSE, organize=TRUE) jesusForData(DT.Spotify_molten) jesusForData(DT.Orchard_molten) jesusForData(DT.Molten) notifyAndEmail("Come Back") ## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## Basic plot of DT.spotify ## atop doesnt work title <- subtext("Ratio of Cumulative Streams over Cumalitive User Counts", sprintf("from %s to date", format.Date(minDate, "%Y-%m"))) title <- paste("Ratio of Cumulative Streams over Cumalitive User Counts", sprintf("from %s to date", format.Date(minDate, "%Y-%m")), sep="\n") levs <- unique(c("US", "DE", "SE", "FR", "MX", "AU", "ES", "DK", "FI", "NL", "NO", DT.Spotify_molten$country)) DT.Spotify_molten[, country := factor(country, levels=levs)] W <- 10 H <- 8 P.Spotify_streams_to_user_ratio <- ggLinegraph(DT.Spotify_molten[product %in% c("P", "A") & country %in% countries_using & upto >= "2013-09-30"], x="upto", y="streams_per_user", color="country", facet_formula="product ~ .", facet_scale="free", title=title) .o(ggsave.out(P.Spotify_streams_to_user_ratio, width=W, height=H)) P.Spotify_streams_to_user_ratio_crop <- ggLinegraph(DT.Spotify_molten[product %in% c("P", "A") & country %in% c("US", "SE", "FI", "DE") & upto >= "2013-09-30"], x="upto", y="streams_per_user", color="country", facet_formula="product ~ .", facet_scale="free", title=title) .o(ggsave.out(P.Spotify_streams_to_user_ratio_crop, width=W, height=H)) P.Spotify_streams_to_user_ratio_us <- ggLinegraph(DT.Spotify_molten[product %in% c("P", "A") & country %in% c("US") & upto >= "2013-09-30"], x="upto", y="streams_per_user", color="country", facet_formula="product ~ .", facet_scale="free", title=title) .o(ggsave.out(P.Spotify_streams_to_user_ratio_us, width=W, height=H)) ggsave.out(plot=list(P.Spotify_streams_to_user_ratio, P.Spotify_streams_to_user_ratio_crop), width=18, height=12, nm="Spotify_Ratio.pdf", outDir="~/Desktop/") ggLinegraph(DT.Orchard_molten[(rank_premium <= 5 | rank_addsupp <= 5)] [product %in% c("P", "A") & country %in% countries_using & upto >= "2013-09-30"], x="upto", y="streams_per_user", color="country", facet_formula="label_name + product ~ .", facet_scale="free") prd <- "P" topX <- 4 DT.plot <- DT.Molten [(rank_premium <= topX | rank_addsupp <= topX)][product == prd][country %in% countries_using & upto >= "2013-09-30"] DT.Molten[owner %in% DT.Molten[rank_premium < 20, unique(owner)], meann(quarterly_change_in_spu), keyby=list(owner, upto)] ggLinegraph(DT.plot, x="upto", y="quarterly_change_in_spu", color="label_name", facet_formula="country ~ .", facet_scale="free") DT.Molten tmp <- DT.Molten[country %in% c("US", "FR")] [labelid %in% c(22389, 11872)] tmp[, quarterly_change_in_spu2 := percentIncrease(streams_per_user), by=setdiff(key(DT.Molten), "upto")] tmp DT.stream_customer_counts_by_label[.("odd", 11872, c("P", "F"), "US")] DT.Molten[.("odd", 11872, c("P", "F"), "US")]