# 2015-07-07 Apple Music Overview.r

# screen -xRR AppleMusic


"
This file unloads the entirety of the raw AppleMusic data and uploads to an S3 bucket, 
then ingests it into snowflake
If then takes the raw spotify data, aggregates it slightly, and sends that to snowflake (via S3)
"


## ---------------------------------------------------------------------------- ##
##   My process is as follows: 
##   
##   Ran one time: 
##   * Create file format
##   * Create @Stage
##   
##   ETL
##   * Extract data from origin, copy to S3 bucket associated to @Stage
##   * Run CREATE OR REPLACE TABLE statement
##   * Run COPY INTO @STAGE statement
##   
## ---------------------------------------------------------------------------- ##


## CRONTAB
# 30 6 * * * Rscript -e 'source("~/git/orch/src/AppleMusic/2015-07-07 Apple Music Export To Snowflake.r")'
# source ("~/git/orch/src/AppleMusic/2015-07-07 Apple Music Export To Snowflake.r")


if (FALSE)
{
  setGitBranchToSystem()
  .g()

  if (FALSE)
  source ("~/git/orch/src/AppleMusic/2015-07-07 Apple Music Export To Snowflake.r")
}


## PARAMETERS CODED INTO THE STAGE
## --------------------------------------------------------- ##
## If these are changed, then the stage & file_format must be recreated
    delim <- "\v"
    use_gzip <- TRUE
## --------------------------------- ##


setScience("AppleMusic", subProj="ETLtoSnowflake", create=TRUE, subl=FALSE, load=FALSE)
.us()

## logging
sinkfile <- sinkOn()$filename

tbl.apple <- "staging_raw_apple_music"
tbl.spot <- c("spotify_streams_cropped_to_apple" = "staging_raw_spotify_v2")
schema     <- "production"
schema_out <- "AppleMusic"
cluster <- 4
wh <- getWH_by_interactive()
dbname <- "prod"
options(snowflake_inuse=TRUE)

snowflake_apple <- TRUE
snowflake_spot  <- TRUE

stage_name <- "Dumps_For_AppleMusic_and_Spotify"
format_name <- "Ricks_Default_RedShift_Dump"

## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

# bucket_spotify <- "s3://dev-rsaporta/BI_Tables/Dumps_For_AppleMusic_and_Spotify/SpotifyStreams_20150713_1327_"
# bucket_apple <- "s3://dev-rsaporta/BI_Tables/Dumps_For_AppleMusic_and_Spotify/AppleMusic_20150713_1327_"

## CONFIRM SF is in order
{
  setSnowflake(wh=wh, dbname=dbname, schema=schema_out, start=TRUE)
  stopifnot("Ricks_Default_RedShift_Dump" %>% toupper %in% sfShowFileFormat(justnames=TRUE, wh=wh, dbname=dbname, schema=schema_out))
  stopifnot(stage_name %>% toupper %in% sfShowStages(justnames=TRUE, wh=wh, dbname=dbname, schema=schema_out))
  ## If not exist, see  '~/git/orch/src/Snowflake_Scratch/SF - stages & file format for Apple Music and Spotify [2015-07-07].r'
  ##   USE schema_out not schema
}

if (FALSE) 
{
  sfCreateFormatAndStage(stage_name=stage_name, format_name=format_name, header=FALSE, delimiter=delim, gzip=use_gzip, overwrite=TRUE, quote_string='"', trim_space=FALSE, create_schema=TRUE
  , comment_format="For ingesting data dumped from Redshift using unloadQry() or unloadTableToS3Bucket()"
  , comment_stage="For raw redshift dumps from staging_raw_apple_music and aggregated staging_raw_spotify_v2. Uses File Format Ricks_Default_RedShift_Dump "
  , wh=wh, dbname=dbname, schema=schema_out
  )
}


## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
tstamp <- now()
bucket_base <- sfGetBucketFromStage(stage_name, wh=wh, dbname=dbname, schema=schema_out, verbose=FALSE)
bucket_apple    <- s3_p(bucket_base, 'AppleMusic',     timeStamp(time=tstamp), prefix=timeStamp("AppleMusic", time=tstamp))
bucket_spotify  <- s3_p(bucket_base, 'SpotifyStreams', timeStamp(time=tstamp), prefix=timeStamp("SpotifyStreams"))

## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

options(snowflake_inuse=FALSE)
setDBall(cluster=cluster)

## APPLE -- whole table
unloadTableToS3Bucket(tbl=tbl.apple, bucket=bucket_apple, schema=schema, cluster=cluster, delimiter=delim, gzip=use_gzip, limit=NULL, force=TRUE)

## SPOTIFY -- aggregate
minDate <- qMinDate(tbl.apple, verbose=FALSE)
dateCol.spot <- c(date="download_date")
colsToAgg.spot <- c(streams="*")
colsToPull=c(dateCol.spot, "ISRC", "OS", "DEVICE_TYPE", "USER_COUNTRY", "USER_ACCESS", "USER_TYPE")
## Construct query
Q.spotify_streams_for_applemusic <- makeQry(colsToPull=colsToPull, dateCol=dateCol.spot, minDate=minDate, expand=FALSE, tbl="staging_raw_spotify_v2", schema=schema, colsToAgg=colsToAgg.spot)

## No need to do anything at this point, it will happen downstream
# ## Populte table from query
# sfPopulateTable(qry=Q.spotify_streams_for_applemusic, bucket=bucket_spotify)

## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

## CREATE QRY for table creation
## sample each table
DT_samp.apple <- headDB(tbl=tbl.apple, schema=schema, cluster=cluster, n=500, snowflake_inuse=snowflake_apple)
DT_samp.spot  <- headDB(tbl=tbl.spot,  schema=schema, cluster=cluster, n=500, cols=colsToPull, snowflake_inuse=snowflake_spot)

## Clean up spotify sample by adding a stream col (with large numeric) and converting posix to date for download_date
DT_samp.spot[, names(colsToAgg.spot) := 1e9] ## add in count information
DT_samp.spot[, (dateCol.spot) := lapply(.SD, as.Date), .SDcols=dateCol.spot]

make_string_col_long_ <- . %>% {suppressWarnings(.[1, nwhich(sapply(., function(x) is.character(x) && lunique(nchar(x)) > 25)) := pasteC(rep("A", 1999))])}
DT_samp.apple %>% make_string_col_long_ %>% {invisible(.)}
DT_samp.spot %>% make_string_col_long_ %>% {invisible(.)}

saveImageTo()
## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

# 2015-08-15 -- I Think the crash is resolved
if (FALSE)
  message(" ~~~~~~~~~~ ____ HERE IS WHERE THE CRASH IS ___________  ~~~~~~~\nright before calling setSnowflake() ....... \n")

## Need a better command to switch to snowflake
dbDisconnectAll()
setSnowflake(wh=wh, dbname=dbname, showWarnings=FALSE)

tbl_out.apple <- "AppleMusicRaw"
tbl_out.spot  <- "SpotifyAggd"

## CREATE THE TABLES
Qry_create.apple  <- makeSQLtable(DT_samp.apple, table.name=tbl_out.apple, wh=wh, dbname=dbname, schema=schema_out, replace=TRUE, diststyle=NULL, sortkey=NULL, snowflake_inuse=TRUE, comment="Raw dump from RedShift", create_only=TRUE,  quiet=FALSE)
Qry_create.spot   <- makeSQLtable(DT_samp.spot,  table.name=tbl_out.spot,  wh=wh, dbname=dbname, schema=schema_out, replace=TRUE, diststyle=NULL, sortkey=NULL, snowflake_inuse=TRUE, comment="Aggregated from staging_raw_spotify_v2", create_only=TRUE,  quiet=FALSE)

## CREATE THE SCHEMA IF NOT EXIST
if (!(sfQry("SHOW SCHEMAS", verbose=FALSE, dbname=dbname) %>% {toupper(schema_out) %in% .$name}))
  paste0("CREATE SCHEMA ", dbschematbl(dbname=getSnowflakeDB(), schema=schema_out)) %>% sfQry(verbose=FALSE)

## SET TO SNOWFLAKE
setSnowflake(wh=wh, dbname=dbname, schema=schema_out)

## CREATE OR REPLACE TABLES
sfQry(Qry_create.apple, wh=wh, dbname=dbname, schema=schema_out, verbose=TRUE)
sfQry(Qry_create.spot,  wh=wh, dbname=dbname, schema=schema_out, verbose=TRUE)

stopifnot(c(tbl_out.apple, tbl_out.spot) %>% toupper %in% sfShowTables(justnames=TRUE, wh=wh, dbname=dbname, schema=schema_out, verbose=FALSE))

## The Stage and Filetype for the data load was created in the public schema
setSnowflake(wh=wh, dbname=dbname, schema=schema_out)
# ret.load_okay.apple    <- sfPopulateTableFromBucket(tbl=tbl_out.apple, schema=schema_out, stage=stage_name, bucket=bucket_apple,   validation_mode=FALSE)
# ret.load_okay.spotify  <- sfPopulateTableFromBucket(tbl=tbl_out.spot,  schema=schema_out, stage=stage_name, bucket=bucket_spotify, validation_mode=FALSE)

ret.load_okay.apple    <- sfPopulateTable(tbl=tbl_out.apple, schema=schema_out, stage_name=stage_name, bucket=bucket_apple,   validation_mode=FALSE)
{
if (snowflake_spot)
  ret.load_okay.spotify  <- sfPopulateTable(qry=Q.spotify_streams_for_applemusic,  schema=schema_out, stage_name=stage_name, validation_mode=FALSE, overwrite=TRUE)
else
  ret.load_okay.spotify  <- sfPopulateTable(tbl=tbl_out.spot, schema=schema_out, stage_name=stage_name, bucket=bucket_spotify,   validation_mode=FALSE)
}

cat("Twelve rows in Apple Music: \n"); print(headDB(tbl=tbl_out.apple, schema=schema_out, n=12, msg_sf=FALSE))
cat("Twelve rows in Spotify: \n"); print(headDB(tbl=tbl_out.spot,  schema=schema_out, n=12, msg_sf=FALSE))

qRowCount(tbl=tbl_out.apple, schema=schema_out, msg_sf=FALSE) %>% formnumb %>% cat("AppleMusic rows: ", ., "\n")
qRowCount(tbl=tbl_out.spot,  schema=schema_out, msg_sf=FALSE) %>% formnumb %>% cat("Spotify rows: ", ., "\n")

# confirm max dates
qMaxDate(tbl=tbl_out.apple, schema=schema_out, where=sprintf("datestamp < '%s'", today()), verbose=FALSE) %>% dateStamp(date=., human=TRUE) %>% cat("AppleMusic maxDae: ", ., "\n")
qMaxDate(tbl=tbl_out.spot, schema=schema_out, verbose=FALSE) %>% dateStamp(date=., human=TRUE) %>% cat("Spotify maxDae: ", ., "\n")

try({
  verboseMsg(verbose, "Creating applemusic_view")
  sfQry("
  CREATE OR REPLACE VIEW applemusic.applemusic_view AS
    SELECT 
      raw.*, 
      rates.rate_per_play, 
      rates.rate_per_play_usd,
      rates.currency,
      1.0 * raw.quantity * rates.rate_per_play_usd * CASE WHEN raw.membership_mode = 'PAID' THEN 4.0 ELSE 1 END AS estimated_gross_usd,
      1.0 * raw.quantity * rates.rate_per_play_usd * CASE WHEN raw.membership_mode = 'PAID' THEN 3.0 ELSE 1 END AS estimated_gross_usd_low,
      1.0 * raw.quantity * rates.rate_per_play_usd * CASE WHEN raw.membership_mode = 'PAID' THEN 5.0 ELSE 1 END AS  estimated_gross_usd_high
    FROM applemusic.applemusicraw raw
    LEFT JOIN applemusic.apple_music_rates rates
    ON raw.storefront_name = rates.country_code
    ")
})

  # NOT GOOD FOR NOW .... #    try({
  # NOT GOOD FOR NOW .... #      verboseMsg(verbose, "Creating applemusic_view")
  # NOT GOOD FOR NOW .... #      sfQry("
  # NOT GOOD FOR NOW .... #      CREATE OR REPLACE VIEW applemusic.applemusic_with_estimates_view AS
  # NOT GOOD FOR NOW .... #        SELECT 
  # NOT GOOD FOR NOW .... #          raw.*, 
  # NOT GOOD FOR NOW .... #          rates.effective_monthly_usd_per_unit_by_country_code, 
  # NOT GOOD FOR NOW .... #          1.0 * raw.quantity * rates.effective_monthly_usd_per_unit_by_country_code AS estimated_gross_usd
  # NOT GOOD FOR NOW .... #        FROM applemusic.applemusicraw raw
  # NOT GOOD FOR NOW .... #        LEFT JOIN (
  # NOT GOOD FOR NOW .... #            SELECT 
  # NOT GOOD FOR NOW .... #            -- bi_accounting.activity_month,
  # NOT GOOD FOR NOW .... #            bi_accounting.country_code,
  # NOT GOOD FOR NOW .... #            CASE WHEN bi_accounting.transac_type_abbr = 'S' THEN 'PAID' WHEN bi_accounting.transac_type_abbr = 'AS' THEN 'TRIAL' ELSE 'OTHER' END AS membership_mode,
  # NOT GOOD FOR NOW .... #            COALESCE(SUM(bi_accounting.gross_revenue_usd),0) / NULLIF(COALESCE(SUM(bi_accounting.units),0), 0) AS effective_monthly_usd_per_unit_by_country_code
  # NOT GOOD FOR NOW .... #            FROM prod.bi.accounting AS bi_accounting
  # NOT GOOD FOR NOW .... #            WHERE 
  # NOT GOOD FOR NOW .... #              (bi_accounting.activity_month >= '2015-06-01') AND
  # NOT GOOD FOR NOW .... #              (bi_accounting.transac_type_abbr in ('S', 'AS')) AND
  # NOT GOOD FOR NOW .... #              (bi_accounting.storeid = 1)
  # NOT GOOD FOR NOW .... #            -- GROUP BY 1,2,3
  # NOT GOOD FOR NOW .... #            GROUP BY 1,2
  # NOT GOOD FOR NOW .... #          ) rates
  # NOT GOOD FOR NOW .... #        ON (raw.storefront_name = rates.country_code AND 
  # NOT GOOD FOR NOW .... #           raw.membership_mode = rates.membership_mode)
  # NOT GOOD FOR NOW .... #        ")
  # NOT GOOD FOR NOW .... #      })


sinkOff()

##         END OF ETL
## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~


## No need to update Looker files
if (FALSE)
try ({
  srcOther("Looker") %>% sourceSupportFns
  file.lookml <- c(tbl_out.apple, tbl_out.spot) %>% sapply(create_lookml_from_tbl, schema=schema_out, wh=wh, dbname=dbname, TEST_RUN=FALSE)
  quickEmail(getRS("to"), files=file.lookml, info=sprintf("LOOKML file for updated %s and %s", schemaPaste(schema=schema_out, tbl_out.apple), schemaPaste(schema=schema_out, tbl_out.spot)))
})





## OBSERVATIONS
## ----------------------------------------------- ##
if (FALSE) {
  out <- sprintf("SELECT count(DISTINCT ISRC) FROM %s.%s", schema_out, tbl_out.spot)
  where <- sprintf("A WHERE A.isrc in (SELECT isrc FROM %s.%s)", schema_out, tbl_out.apple)
  sprintf("SELECT ((%s %s) / (%1$s)) AS Percent_of_ISRCs_Apple_has_in_spotify", out, where) %>% sfQry
  ret_where <- sprintf("(%s %s)", out, where) %>% sfQry
  ret_out <- sprintf("(%s)", out) %>% sfQry
  fwp(502995 / 1100096)
  fwp(502995 / 3411109)

  4711962
  934240
}

# 
# Unique tracks in Spotify == 3,411,109
# Unique tracks in Music  == 1,100,096
# 
# Percent of unique Tracks in Spotify that are in Music : 14.8 %
# Percent of unique Tracks in Music that are in Spotify : 45.7 %
# * by "in" we mean streamed through that service

## ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~