#Dependencies import os # comment when running locally import subprocess subprocess.check_call('pip install -r /opt/ml/processing/input/dependencies/requirements.txt', shell=True) import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', ) import random import pandas as pd import boto3 import math import random import scipy import numpy as np import statsmodels.formula.api as smf import statsmodels.api as sm import pmdarima as pm import time from datetime import datetime, date, timedelta from statsmodels.tsa.arima.model import ARIMA from statsmodels.tools.eval_measures import mse,rmse from statsmodels.tsa.statespace.tools import diff from scipy import fftpack, signal from multiprocessing import Pool, cpu_count def timing_val(func): ''' Utility function that calculates the execution time of other functions. When @timing_val is present, the function will return a tuple containing result and execution time. From source: http://www.daniweb.com/code/snippet368.html ''' def wrapper(*arg, **kw): t1 = time.time() res = func(*arg, **kw) t2 = time.time() return (t2 - t1), res, func.__name__ return wrapper ################################################################################################################## ############################################# Lags Analysis ###################################################### ################################################################################################################## def iterate_group_by_key(df, col_for_key, sort_data=True): ''' Optimized function (generator) that iterates over the DataFrame and groups it by a specified column. It operates on a sorted dataframe, so all rows for a given key (here ISRC) are together. It yields one group at a time as a new DataFrame. ''' if sort_data: df = df.sort_values(by=col_for_key) # helper function that returns the value in the column specified by col_for_key at the index i. def key_at(i): return np.array(df[col_for_key].iloc[i]) index, size = (0, df.shape[0]) while index < size: current_key = key_at(index) res = [] while index < size and list(current_key) == list(key_at(index)): res.append(df.iloc[index]) index = index + 1 resdf = pd.DataFrame(res, columns=df.columns) yield resdf @timing_val def lag_calculation(df): new_df = pd.DataFrame(columns=['ISRC','lag_creations','lag_views']) count = 1 for subdf in iterate_group_by_key(df, ['ISRC']): df_isrc = subdf isrc = df_isrc['ISRC'].iloc[0] df_isrc = df_isrc.sort_values(by=['ISRC']) try: # Creations corr_creations = signal.correlate(df_isrc[['TIKTOK_CREATIONS']] - np.mean(df_isrc[['TIKTOK_CREATIONS']]), df_isrc[['STREAMS']] - np.mean(df_isrc[['STREAMS']]), method= 'auto', mode='full') lags_creations = signal.correlation_lags(len(df_isrc[['TIKTOK_CREATIONS']]), len(df_isrc[['STREAMS']]), mode='full') lag_creations = lags_creations[np.argmax(abs(corr_creations))] # Views corr_views = signal.correlate(df_isrc[['TIKTOK_VIDEO_VIEWS']] - np.mean(df_isrc[['TIKTOK_VIDEO_VIEWS']]), df_isrc[['STREAMS']] - np.mean(df_isrc[['STREAMS']]), method= 'auto', mode='full') lags_views = signal.correlation_lags(len(df_isrc[['TIKTOK_VIDEO_VIEWS']]), len(df_isrc[['STREAMS']]), mode='full') lag_views = lags_views[np.argmax(abs(corr_views))] new_df = pd.concat([new_df, pd.DataFrame([{ 'ISRC': isrc, 'lag_creations': lag_creations, 'lag_views': lag_views }], columns=new_df.columns)]) if count % 100 ==0: print("Worker processed {} ISRCs.".format(count)) count = count+1 except Exception as e: print("Error while processing ISRC {}: '{}'".format(isrc, e)) return new_df def save_chunk_s3(df, run_id, chunk_id, dry_run): '''Save the given parallel chunk result into S3''' if dry_run: logging.warning(f"DRY_RUN flag is True, no parallel chunk will be saved.") else : s3 = boto3.client('s3') bucket_name = 'dev-cucumbers' today = datetime.today().strftime('%Y%m%d-%H%M%S') music_genre = 'Hip-hop_Rap' filepath = "eimpara/TikTok_analysis/Lags/Lags_{}_{}_{}_{}.csv".format(music_genre, today, chunk_id, run_id) csv_buffer = df.to_csv(index=False).encode('utf-8') s3.put_object(Body=csv_buffer, Bucket=bucket_name, Key=filepath) logging.info(f"Table saved to S3 bucket: {bucket_name}, with file name: {filepath}") ############################################# UTILS ############################################################## ################################################################################################################## def check_one_track_df(one_track_df): '''Checking that the length of the individual time series is 140 days''' return len(one_track_df)==140 def unique_isrc_df(isrc, df): ''' As time series analysis is univariate, decomposing the dataframe as to have one-track dataframes at time of analysis. ''' one_track_df = df[df['ISRC']==isrc].copy() one_track_df['ACTIVITY_DATE'] = pd.to_datetime(one_track_df['ACTIVITY_DATE']) one_track_df['ACTIVITY_DATE'] = one_track_df['ACTIVITY_DATE'].dt.date one_track_df = one_track_df[['ISRC', 'ACTIVITY_DATE', 'STREAMS']] return one_track_df.sort_values(by = 'ACTIVITY_DATE') def data_for_chunk(df, chunk_id, chunk_count): ''' Function sorting ISRCs and splitting them into n chuncks and taking only one of them (chunk_id) ''' IDs = df['ISRC'].unique().tolist() IDs.sort() chunk_size = len(IDs) // chunk_count chunked = [IDs[n:n+chunk_size] for n in range(0, len(IDs), chunk_size)] ids = chunked[chunk_id] chunk_df = df[df['ISRC'].isin(ids)] return chunk_df def split_universe(df, n): ''' Function splitting dataframe for parallelism df -- initial dataframe n -- number of chunks ''' # list of sorted ISRCs into n chunks def chunks(l, n): for i in range(0, len(l), n): yield l[i:i + n] isrcs = df['ISRC'].unique().tolist() isrcs.sort() isrc_chunks = chunks(isrcs, (len(isrcs) // n) + 1) return list(isrc_chunks) @timing_val def split_work(full_df, n, f, dry_run): ''' full_df -- full dataframe that will be split into n parallel tasks n -- number of parallel tasks f -- function handling one parallel task dry_run -- bolean flag indicating whether the results are saved. ''' isrc_chunks = split_universe(full_df, n) splitted_df = [] # For each chunk, creating an individual dataframe that only includes rows with the same ID for chunk in isrc_chunks: df_chunk = full_df[full_df['ISRC'].isin(chunk)] splitted_df.append(df_chunk) # Unique run ID generator based on timestamp # This is to allow to stitch back files together run_id = str(time.time() * 1000000) logging.info("ISRCs have been splitted in {} dataframes for run id={}".format(n, run_id)) # Using multiprocessing pool to apply the function to each chunk with Pool(n) as p: # Adding indices to the list of chunks splitted_df_with_index = list(enumerate(splitted_df)) # List of arguments for the function splitted_df_with_args = map(lambda x: (dry_run, run_id, x[0], x[1]), splitted_df_with_index) p.map(f, splitted_df_with_args) logging.info("Moments calculation finished.") def work_load(arg): ''' Function that returns the result of arima_iscrs() and saves it in S3''' (dry_run, run_id, chunk_id, df) = arg logging.info(f"Arguments {run_id},{chunk_id},{type(df)}") timing, res, _ = lag_calculation(df) save_chunk_s3(res, run_id, chunk_id, dry_run) return res @timing_val def load_full_data(df): subset_for_pred = df[df['GENRENAME']=='Hip-hop/Rap'] return data_for_chunk(subset_for_pred, CHUNK_ID, CHUNK_COUNT) # list_rock = subset_for_pred['ISRC'].unique().tolist() # sample_size = 10 # random_sample = random.sample(list_rock, k=sample_size) # rock_subset = subset_for_pred[subset_for_pred['ISRC'].isin(random_sample)].copy() # return data_for_chunk(rock_subset, CHUNK_ID, CHUNK_COUNT) if __name__ == '__main__': # Dry run = True => data will NOT be saved DRY_RUN = False # Splits the universe of ids into CHUNK_COUNT stable chunks CHUNK_COUNT = 1 # We are running the script only for the below CHUNK_ID, out of CHUNK_COUNT chunks CHUNK_ID = 0 # Data CSV File INPUT_FILE = 's3://dev-cucumbers/eimpara/TikTok_analysis/genre_level_analysis.20231204-145402.csv' # Number of parallel processes to process the data PARALLELISM = 30 #cpu_count() #max(1, cpu_count() // 2) # VAR_EXOG = 'TIKTOK_VIDEO_VIEWS' # MUSIC_GENRE = 'Hip-hop/Rap' if DRY_RUN: logging.warning(f"DRY_RUN flag is True, no data will be saved !!!") logging.info(f"Reading data from CSV '{INPUT_FILE}'") logging.info(f"Computing chunk {CHUNK_ID} out of {CHUNK_COUNT} chunk(s)") logging.info(f"Data will be processed with parallelism={PARALLELISM}") # Reading input data file from S3 df = pd.read_csv(INPUT_FILE) # Fixing dates that are not read with the correct type in a CSV file df['ACTIVITY_DATE'] = pd.to_datetime(df['ACTIVITY_DATE']) df['ACTIVITY_DATE'] = df['ACTIVITY_DATE'].dt.date logging.info(f"Min activity date: {df['ACTIVITY_DATE'].min()}") logging.info(f"Max activity date: {df['ACTIVITY_DATE'].max()}") # Computing ARIMA for all filtered ids from CHUNK_ID parallelism = PARALLELISM timing, full_df, _ = load_full_data(df) logging.info(f"{full_df.shape[0]} rows loaded in {timing} seconds") logging.info(f"Splitting work into {parallelism} parts for parallel processing") elapsed_time = split_work(full_df, parallelism, work_load, DRY_RUN) logging.info(f"Data processed in {elapsed_time} seconds")