{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Resources\n- https://github.com/otto-de/recsys-dataset\n- https://www.kaggle.com/competitions/otto-recommender-system/discussion/364534\n\ncreate datasets to train on and for evaluation -> train MF/word2vec -> create a covisitation matrix (separate for train with validation and for the full train set for submission) -> create features & diagnostic code (measure hit rate, recall) for ranking models -> train a ranking model -> create submission based on this output","metadata":{}},{"cell_type":"markdown","source":"# Setup","metadata":{}},{"cell_type":"markdown","source":"## Import libraries","metadata":{}},{"cell_type":"code","source":"import os\n\ntry: \n    import pyspark\nexcept ModuleNotFoundError:\n    os.system(\"pip install pyspark\")\n\nimport psutil\nimport shutil\n\nimport numpy  as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\nfrom datetime import timedelta\n\nfrom typing import Tuple\n\nfrom pyspark.sql import SparkSession, functions as f, DataFrame\nfrom pyspark.sql.types import DoubleType, ArrayType, IntegerType, StructType, DateType\nfrom pyspark.sql.types import StringType, StructField\nfrom pyspark.sql.window import Window\n\nimport matplotlib.pyplot as plt\n\nimport random","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:38:12.881809Z","iopub.execute_input":"2023-01-25T06:38:12.882750Z","iopub.status.idle":"2023-01-25T06:39:00.241727Z","shell.execute_reply.started":"2023-01-25T06:38:12.882636Z","shell.execute_reply":"2023-01-25T06:39:00.240679Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"pd.set_option('display.max_colwidth', 1000) # Display characters in Pandas df before truncating","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:00.243281Z","iopub.execute_input":"2023-01-25T06:39:00.243664Z","iopub.status.idle":"2023-01-25T06:39:00.250468Z","shell.execute_reply.started":"2023-01-25T06:39:00.243581Z","shell.execute_reply":"2023-01-25T06:39:00.249017Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## User Input","metadata":{}},{"cell_type":"code","source":"# Set random seed\nv_seed = 423\nrandom.seed(v_seed)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:00.288324Z","iopub.execute_input":"2023-01-25T06:39:00.288969Z","iopub.status.idle":"2023-01-25T06:39:00.298504Z","shell.execute_reply.started":"2023-01-25T06:39:00.288933Z","shell.execute_reply":"2023-01-25T06:39:00.297292Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Create Spark session","metadata":{}},{"cell_type":"code","source":"total_memory = int(np.round(0.000000001 * psutil.virtual_memory().total,0)) # convert to GB\nprint(f'\\nTotal available memory: {total_memory}GB')\n\n# This allocates three quarters of the instance's total memory to the driver:\nallocate_to_driver = int(np.round((3/4)*total_memory,0))\nprint(f'Allocate to driver: {allocate_to_driver}GB\\n The remaining memory can be used by the additional nodes/executors.')\n\n# Create a SparkSession\nspark = SparkSession \\\n    .builder\\\n    .config(\n        \"spark.driver.memory\", \n        f'{allocate_to_driver}G')\\\n    .config(\n        \"spark.driver.maxResultSize\",\n        \"0\")\\\n    .getOrCreate()","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:00.299905Z","iopub.execute_input":"2023-01-25T06:39:00.300250Z","iopub.status.idle":"2023-01-25T06:39:06.525316Z","shell.execute_reply.started":"2023-01-25T06:39:00.300218Z","shell.execute_reply":"2023-01-25T06:39:06.523900Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Import functions","metadata":{}},{"cell_type":"code","source":"def add_datetime_min(data: DataFrame) -> DataFrame:\n    \"\"\"\n    :output: DataFrame with added column `datetime_min`\n    \"\"\"\n    w    = Window.partitionBy(\"session\")\n    data = data.withColumn('datetime_min', f.min('datetime').over(w))\n    return data\n\ndef add_datetime_max(data: DataFrame) -> DataFrame:\n    \"\"\"\n    :output: DataFrame with added column `datetime_max`\n    \"\"\"\n    w    = Window.partitionBy(\"session\")\n    data = data.withColumn('datetime_max', f.max('datetime').over(w))\n    return data","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.527020Z","iopub.execute_input":"2023-01-25T06:39:06.527469Z","iopub.status.idle":"2023-01-25T06:39:06.538044Z","shell.execute_reply.started":"2023-01-25T06:39:06.527419Z","shell.execute_reply":"2023-01-25T06:39:06.536605Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def f_non_overlapping_sets_of_sessions(train_df: DataFrame, \n                                       start_dt_valid : pd._libs.tslibs.timestamps.Timestamp, \n                                       seed: int) -> Tuple[DataFrame]:\n    \n    \"\"\"\n    This function creates a non-overlapping set of profile_id`s to be used for the training, validation and test sets respectively. \n    The validation set contains sessions that satisfy: `datetime_min >= v_start_dt_valid`. In other words, it only contains sessions that started\n    during the validation set date-range.\n    \n    :param train_df: \n    :param start_dt_valid: \n    :param seed: int representing the seed for random operations\n    \n    :return: Tuple[DataFrame] of 2 DataFrames contaning a non-overlapping set of profile_id`s to be used for the training, validation sets respectively.\n    \"\"\"\n    \n    all_session_ids = train_df.select('session').distinct().persist()\n\n    print(f'all_session_ids = {all_session_ids.count():,}')\n\n    valid_session_ids = train_df\\\n        .where(f.col('datetime_min') >= v_start_dt_valid)\\\n        .select('session')\\\n        .distinct()\\\n        .persist()\n\n    print(f'valid_session_ids = {valid_session_ids .count():,}')\n\n    train_session_ids  = all_session_ids\\\n        .alias('x')\\\n        .join(valid_session_ids.alias('y'), on='session', how='left')\\\n        .where(f.col('y.session').isNull())\\\n        .select('x.session')\\\n        .distinct()\\\n        .persist()\n\n    print(f'train_session_ids = {train_session_ids .count():,}')    \n    \n    return train_session_ids, valid_session_ids","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.539848Z","iopub.execute_input":"2023-01-25T06:39:06.540328Z","iopub.status.idle":"2023-01-25T06:39:06.711500Z","shell.execute_reply.started":"2023-01-25T06:39:06.540284Z","shell.execute_reply":"2023-01-25T06:39:06.710196Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def create_sampled_table(dataframe: DataFrame, fraction: float, seed: float = v_seed) -> DataFrame:\n    \"\"\"\n    :param dataframe: a dataframe that contains the `session` column which this function samples\n    :param fraction: percentage % of the total numnber of sessions in `dataframe` to sample\n    :param seed: seed for the random sampling \n    \n    :return: returns `dataframe` but with sampled sessions based on given `fraction`\n    \"\"\"\n    \n    distinct_sessions = dataframe.select('session').distinct()\n\n    sampled_sessions  = distinct_sessions.sample(withReplacement=False, fraction=fraction, seed=seed).cache()\n\n    total_sessions_cnt   = dataframe.select('session').distinct().count()\n    sampled_sessions_cnt = sampled_sessions.count()\n\n    print(f\"\"\"\n    Total sessions     = {total_sessions_cnt:,}\n    Sampled sessions   = {sampled_sessions_cnt:,}\n    % Sampled sessions = {100*fraction}%\n    \"\"\")\n\n    dataframe_sampled = dataframe.alias('x')\\\n        .join(sampled_sessions.alias('y'), on='session', how='inner')\\\n        .select('x.*')\\\n        .withColumn('date', f.to_date('datetime'))\\\n        .orderBy(['session','datetime'])\n\n    return(dataframe_sampled)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.713470Z","iopub.execute_input":"2023-01-25T06:39:06.714168Z","iopub.status.idle":"2023-01-25T06:39:06.726524Z","shell.execute_reply.started":"2023-01-25T06:39:06.714123Z","shell.execute_reply":"2023-01-25T06:39:06.725575Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\ndef f_train_valid_split(train_df       : DataFrame,\n                        start_dt_train : pd._libs.tslibs.timestamps.Timestamp,\n                        end_dt_train   : pd._libs.tslibs.timestamps.Timestamp,\n                        start_dt_valid : pd._libs.tslibs.timestamps.Timestamp,\n                        end_dt_valid   : pd._libs.tslibs.timestamps.Timestamp,\n                        train_session_ids: DataFrame,\n                        valid_session_ids: DataFrame\n                       \n                      ) -> Tuple[DataFrame]:\n    \"\"\"   \n    \"\"\"\n    \n    # Split train and valid sets by selected date ranges & non-overlapping sets of session id`s\n    train = train_df\\\n        .where((f.col(\"datetime_ref\") >= start_dt_train) & (f.col(\"datetime_ref\") <= end_dt_train))\\\n        .alias('x')\\\n        .join(train_session_ids.alias('y'), on='session', how='inner')\\\n        .orderBy('session','datetime')\\\n        .select('x.*')\n\n    valid = train_df\\\n        .where((f.col(\"datetime_ref\") >= start_dt_valid) & (f.col(\"datetime_ref\") <= end_dt_valid))\\\n        .alias('x')\\\n        .join(valid_session_ids.alias('y'), on='session', how='inner')\\\n        .orderBy('session','datetime')\\\n        .select('x.*')\n\n    ####################################################################################################\n    # Separate ground truth / labels (i.e. what actually happened in the future of datetime_ref) to available historical info\n    ####################################################################################################\n    v_cols = ['session','datetime','aid','type', 'datetime_ref']\n    \n    train_labels = train.where(f.col('datetime') >  f.col('datetime_ref')).select(*v_cols)\n    valid_labels = valid.where(f.col('datetime') >  f.col('datetime_ref')).select(*v_cols)\n\n    train              = train.where(f.col('datetime') <= f.col('datetime_ref')).select(*v_cols)\n    valid              = valid.where(f.col('datetime') <= f.col('datetime_ref')).select(*v_cols)\n    \n    ####################################################################################################\n    # Bring the _labels datasets into the appropriate format (i.e. one row per session per type, with associated aid`s concatenated in python lists)\n    ####################################################################################################\n    v_cols = ['session','type', 'ground_truth']\n    \n    # For each session and type, concatenate all aid`s in a single python list. When type='clicks', only keep the first click for the ground truth:\n    train_labels = train_labels\\\n        .groupby('session','type')\\\n        .agg(f.collect_list(\"aid\").alias('aid_list'))\\\n        .withColumn(\"ground_truth\", f.when((f.col('type')=='clicks'), first_element_udf(\"aid_list\")).otherwise(all_elements_udf(\"aid_list\")) )\\\n        .select(*v_cols)\\\n        .orderBy('session','type')\n    \n    valid_labels = valid_labels\\\n        .groupby('session','type')\\\n        .agg(f.collect_list(\"aid\").alias('aid_list'))\\\n        .withColumn(\"ground_truth\", f.when((f.col('type')=='clicks'), first_element_udf(\"aid_list\")).otherwise(all_elements_udf(\"aid_list\")) )\\\n        .select(*v_cols)\\\n        .orderBy('session','type')\n    \n    \n    return train, train_labels, valid, valid_labels","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.727981Z","iopub.execute_input":"2023-01-25T06:39:06.728523Z","iopub.status.idle":"2023-01-25T06:39:06.746083Z","shell.execute_reply.started":"2023-01-25T06:39:06.728491Z","shell.execute_reply":"2023-01-25T06:39:06.744938Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def add_reference_dt(data: DataFrame, isTest: bool = False) -> DataFrame:\n    \"\"\"\n    Adds to the dataframe a column `datetime_ref` which represents a randomly chosen reference datetime for the given `session`.    \n    :param data: DataFrame would be one of train/validation/test sets, which are required to have the following cols: a) session, b) datetime\n    \n    :output: DataFrame is same as `data` but with added column `datetime_ref`\n    \"\"\"\n    \n    if isTest:\n        \n        # If actual test set, datetime_ref would simply correspond to the datetime_max:\n        data = add_datetime_max(data)\n        data = data.withColumnRenamed('datetime_max', 'datetime_ref')\n        \n    else:\n        # 0. Get min/max datetimes for each session\n        data = add_datetime_min(data)\n        data = add_datetime_max(data)\n\n        # 1. Define a user-defined function (UDF) that generates a random float value between 0 and 1: \n        random_float_udf = f.udf(random.random, returnType=DoubleType()).asNondeterministic()\n        \n        # 2. Use the random_float_udf function and the when function to create the new column datetime_ref:\n        sessions = data.select('session','datetime_min','datetime_max').distinct()\n        \n        sessions = sessions\\\n            .withColumn('random_float', random_float_udf())\\\n            .withColumn('datetime_ref', \n                         f.from_unixtime( f.unix_timestamp('datetime_min') + \n                                            (f.unix_timestamp('datetime_max') - f.unix_timestamp('datetime_min')) * f.col('random_float')\n                                        )\n                        )\\\n            .select('session','datetime_ref')\\\n            .cache()\n        \n        # 3. Add `datetime_ref` to main dataframe:\n        data = data\\\n            .drop('datetime_ref')\\\n            .alias('x')\\\n            .join(sessions.alias('y'), on='session', how='left')\\\n            .select('x.*','y.datetime_ref')\n    \n    return(data)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.750307Z","iopub.execute_input":"2023-01-25T06:39:06.750941Z","iopub.status.idle":"2023-01-25T06:39:06.762624Z","shell.execute_reply.started":"2023-01-25T06:39:06.750890Z","shell.execute_reply":"2023-01-25T06:39:06.761176Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def add_session_type(data: DataFrame) -> DataFrame:\n    \"\"\"\n    :output: DataFrame with added column `session_type`\n    \"\"\"\n    data = data.withColumn('session_type', f.concat(f.col('session'),f.lit('_'),f.col('type')))\n    return data","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.764515Z","iopub.execute_input":"2023-01-25T06:39:06.765067Z","iopub.status.idle":"2023-01-25T06:39:06.779329Z","shell.execute_reply.started":"2023-01-25T06:39:06.765021Z","shell.execute_reply":"2023-01-25T06:39:06.778233Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def create_predictions_df(data: DataFrame) -> DataFrame:\n    \"\"\"\n    :return: a DataFrame that contains a cross-join of all unique sessions in `data` and the 3 action types in 'clicks', 'carts', 'orders'. \n             Returned dataframe has three columns: [`session_type`, `session`, `type`]\n    \"\"\"\n    \n    # Create unique sessions (as PySpark df)\n    sessions = data.select('session').distinct()\n\n    # Create unique types (as PySpark df)\n    schema = StructType( [StructField(\"type\", StringType(), True)] )\n    types  = spark.createDataFrame(pd.DataFrame(['clicks', 'carts', 'orders']), schema)\n\n    # CrossJoin to get all combos (as PySpark df):\n    predictions = sessions.crossJoin(types).orderBy('session','type')\n    predictions = add_session_type(predictions).select('session_type','session','type')\n    \n    return(predictions)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.780704Z","iopub.execute_input":"2023-01-25T06:39:06.781131Z","iopub.status.idle":"2023-01-25T06:39:06.795934Z","shell.execute_reply.started":"2023-01-25T06:39:06.781094Z","shell.execute_reply":"2023-01-25T06:39:06.794701Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Define the transformation function\ndef first_element(lst:list)->list:\n    return lst[0:1]\n\ndef all_elements(x):\n    return x\n\n\n# When VALIDATING we need to be able to f.explode() the column created by the below UDFs. That's only possible if its values are of ArrayType rather \n# than strings.\nfirst_element_udf = f.udf(first_element, returnType=ArrayType(IntegerType()))\nall_elements_udf  = f.udf(all_elements,  returnType=ArrayType(IntegerType()))\n","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.797199Z","iopub.execute_input":"2023-01-25T06:39:06.798150Z","iopub.status.idle":"2023-01-25T06:39:06.807908Z","shell.execute_reply.started":"2023-01-25T06:39:06.798114Z","shell.execute_reply":"2023-01-25T06:39:06.806889Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Takes a Pyspark df's column that contains a list of strings and converts it into a list of integers. \n# This allows us to apply functions first_element_udf & all_elements_udf later on.\nconvert_to_int = f.udf(lambda x: [int(i) for i in x], ArrayType(IntegerType()))","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.809443Z","iopub.execute_input":"2023-01-25T06:39:06.809962Z","iopub.status.idle":"2023-01-25T06:39:06.819191Z","shell.execute_reply.started":"2023-01-25T06:39:06.809918Z","shell.execute_reply":"2023-01-25T06:39:06.818042Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Define the transformation function\ndef extract_type(lst:list):\n    return lst[1]\n\ndef extract_session(lst:list):\n    return lst[0]\n\n# Create a UDF (user-defined function) from the transformation function\nextract_type_udf    = f.udf(extract_type)\nextract_session_udf = f.udf(extract_session)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.823484Z","iopub.execute_input":"2023-01-25T06:39:06.823881Z","iopub.status.idle":"2023-01-25T06:39:06.831819Z","shell.execute_reply.started":"2023-01-25T06:39:06.823829Z","shell.execute_reply":"2023-01-25T06:39:06.830696Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def extract_session_and_type(data: DataFrame) -> DataFrame:\n    \"\"\"\n    Creates two columns `session` and `type` with values extracted by the existing column `session_type`\n    :return: `data` DataFrame but with added columns \"session\" and \"type\"\n    \"\"\"\n    \n    data = data\\\n    .withColumn(\"session\", extract_session_udf(f.split(\"session_type\", \"_\")))\\\n    .withColumn(\"type\",    extract_type_udf(f.split(\"session_type\", \"_\")))\n    \n    return(data)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.833168Z","iopub.execute_input":"2023-01-25T06:39:06.833504Z","iopub.status.idle":"2023-01-25T06:39:06.849465Z","shell.execute_reply.started":"2023-01-25T06:39:06.833473Z","shell.execute_reply":"2023-01-25T06:39:06.848286Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Read-in data","metadata":{}},{"cell_type":"markdown","source":"## Train set","metadata":{}},{"cell_type":"code","source":"train_df = spark.read.parquet(\"/kaggle/input/pre-process-with-pyspark-into-parquet/train/\").orderBy(['session', 'datetime'])\n\nprint(f\"Rows train_df= {train_df.count():,}\\nCols train_df= {len(train_df.columns)}\")\n\ntrain_df.limit(5).toPandas()","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:39:06.851079Z","iopub.execute_input":"2023-01-25T06:39:06.851630Z","iopub.status.idle":"2023-01-25T06:39:56.660746Z","shell.execute_reply.started":"2023-01-25T06:39:06.851593Z","shell.execute_reply":"2023-01-25T06:39:56.659087Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Test set","metadata":{}},{"cell_type":"code","source":"%%time \ntest_df = spark.read.parquet(\"/kaggle/input/pre-process-with-pyspark-into-parquet/test/\").orderBy(['session', 'datetime'])\n\nprint(f\"Rows test_df = {test_df.count():,}\\nCols test_df = {len(test_df.columns)}\")\n\ntest_df.limit(5).toPandas()","metadata":{"execution":{"iopub.status.busy":"2023-01-23T08:14:54.942225Z","iopub.execute_input":"2023-01-23T08:14:54.943218Z","iopub.status.idle":"2023-01-23T08:14:56.416040Z","shell.execute_reply.started":"2023-01-23T08:14:54.943165Z","shell.execute_reply":"2023-01-23T08:14:56.415130Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Sample submission","metadata":{}},{"cell_type":"code","source":"%%time\nsample_submission = spark.read.csv(\"/kaggle/input/otto-recommender-system/sample_submission.csv\", header=True)\n\n# Extract the session and type from string:\nsample_submission = extract_session_and_type(sample_submission)\n\nsample_submission.limit(5).toPandas()","metadata":{"execution":{"iopub.status.busy":"2023-01-23T08:14:56.417212Z","iopub.execute_input":"2023-01-23T08:14:56.417789Z","iopub.status.idle":"2023-01-23T08:14:58.225042Z","shell.execute_reply.started":"2023-01-23T08:14:56.417755Z","shell.execute_reply":"2023-01-23T08:14:58.223850Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Write full datasets as parquet","metadata":{}},{"cell_type":"markdown","source":"### Create a random reference datetime for each session.\nThe train/valid split will be based on this reference dt.\n\n**Reference datetime** = It is a randomly selected datetime for a session, before which (inclusive) all historical data are assumed to be known and can be used for making preedictions, while all historical data after (not inclusive) it are assumed to be \"in the future\" and represent the *ground truth* of what we're trying to predict.\n\nThis logic of separating \"past\" and \"future\" data with respect to the given reference datetime is implemented by the function `f_train_valid_split`, and here you can see the part of is which, for example, creates the ground truth dataset for train:\n- `train_ground_truth = train.where(f.col('datetime') > f.col('datetime_ref')).select(*v_cols1)`\n\nFor the **Test set**, the reference datetime will correspond to the latest date available. For that reason, any actions taken place on `datetime_ref` should be included in the historical data for inference.\n","metadata":{}},{"cell_type":"code","source":"%%time\n# Non-sampled train/test:\n\nv_cols = ['session','datetime','aid','type','datetime_ref']\n\ntrain_df = add_reference_dt(train_df, isTest=False).select(*v_cols)\ntest_df  = add_reference_dt(test_df,  isTest=True) .select(*v_cols)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\n\nTRAIN_FULL_PATH = f\"/kaggle/working/train_full/\"\n\nif not os.path.isdir(TRAIN_FULL_PATH):\n    print(f'Extracting data in path: `{TRAIN_FULL_PATH}`')\n    train_df.write.mode(\"overwrite\").parquet(TRAIN_FULL_PATH)\nelse:\n    print('Dataset has already been produced - not exporting.')","metadata":{"execution":{"iopub.status.busy":"2023-01-23T08:24:27.146530Z","iopub.execute_input":"2023-01-23T08:24:27.147956Z","iopub.status.idle":"2023-01-23T08:24:27.158035Z","shell.execute_reply.started":"2023-01-23T08:24:27.147895Z","shell.execute_reply":"2023-01-23T08:24:27.156281Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\n\nTEST_FULL_PATH = f\"/kaggle/working/test_full/\"\n\nif not os.path.isdir(TEST_FULL_PATH):\n    print(f'Extracting data in path: `{TEST_FULL_PATH}`')\n    test_df.write.mode(\"overwrite\").parquet(TEST_FULL_PATH)\nelse:\n    print('Dataset has already been produced - not exporting.')","metadata":{"execution":{"iopub.status.busy":"2023-01-23T08:24:27.159934Z","iopub.execute_input":"2023-01-23T08:24:27.161863Z","iopub.status.idle":"2023-01-23T08:24:27.171930Z","shell.execute_reply.started":"2023-01-23T08:24:27.161802Z","shell.execute_reply":"2023-01-23T08:24:27.170173Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Create sampled tables train/valid & export as parquet","metadata":{}},{"cell_type":"code","source":"v_checkPointPath = \"/root/tmp/checkpoint_data\"\nspark.sparkContext.setCheckpointDir(v_checkPointPath)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:45:37.072746Z","iopub.execute_input":"2023-01-25T06:45:37.073150Z","iopub.status.idle":"2023-01-25T06:45:37.083500Z","shell.execute_reply.started":"2023-01-25T06:45:37.073117Z","shell.execute_reply":"2023-01-25T06:45:37.082395Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\n\nfor FRACTION in [0.004, 0.05, 0.1, 0.2]:\n    \n    print(f\"\"\"\n    *************************************************************\n    ***************    FRACTION = {FRACTION}    *****************\n    *************************************************************\n    \"\"\")\n    \n    # Take sample:\n    train_dfs = create_sampled_table(train_df, fraction=FRACTION).checkpoint()\n    train_dfs.persist()\n    print(f\"Rows = {train_dfs.count():,}\\nCols = {len(train_dfs.columns)}\")\n\n    \n    # Create train & validation sets\n    ## Get date limits for reference date\n    # Start/End dates of the whole dataset:\n    v_start_dt = pd.to_datetime(train_dfs.agg(f.min('datetime').alias('value')).collect()[0]['value'])\n    v_end_dt   = pd.to_datetime(train_dfs.agg(f.max('datetime').alias('value')).collect()[0]['value'])\n\n    # Start/End dates of each split dataset:\n    v_end_dt_valid    = v_end_dt\n    v_start_dt_valid  = v_end_dt_valid + timedelta(days=-7)\n\n    v_end_dt_train   = v_start_dt_valid + timedelta(minutes=-1)\n    v_start_dt_train = v_start_dt\n\n    assert  v_end_dt_valid > v_start_dt_valid > v_end_dt_train > v_start_dt_train, 'Chosen dates must obey this inequality'\n\n    print(f\"\"\"\n    v_start_dt = {v_start_dt}\n    v_end_dt   = {v_end_dt}\n\n    v_start_dt_train = {v_start_dt_train}\n    v_end_dt_train   = {v_end_dt_train}\\n\n    v_start_dt_valid = {v_start_dt_valid}\n    v_end_dt_valid   = {v_end_dt_valid}\n    \"\"\")\n    \n    ## Define audience for train & validation sets\n    # - The validation set should contain sessions whose:  `datetime_min >= v_start_dt_valid`\n    # **Note**: This limitation is only relevant to this competition, as in a general use-case those sessions could have an available history prior to `v_start_dt_valid`.\n    \n    train_dfs = add_datetime_min(train_dfs)\n\n    train_session_ids, valid_session_ids = f_non_overlapping_sets_of_sessions(train_dfs,\n                                                                              start_dt_valid = v_start_dt_valid,\n                                                                              seed=v_seed)\n    \n\n\n\n    ### Create train/valid sets based on selected date range and non-overlapping sets of sessions\n    # Sampled train/valid:\n    train, train_labels, valid, valid_labels = f_train_valid_split( train_dfs\n                                                                 ,v_start_dt_train\n                                                                 ,v_end_dt_train\n                                                                 ,v_start_dt_valid\n                                                                 ,v_end_dt_valid\n                                                                 ,train_session_ids\n                                                                 ,valid_session_ids\n                                                                  )\n    train.persist()\n    train_labels.persist()\n    valid.persist()\n    valid_labels.persist()   \n    \n    agg_ops =[f.min('datetime')         .alias('datetime_min')\n             ,f.max('datetime')         .alias('datetime_max')\n             ,f.min('datetime_ref')     .alias('datetime_ref_min')\n             ,f.max('datetime_ref')     .alias('datetime_ref_max')\n             ,f.min('session')          .alias('sessionid_min')\n             ,f.max('session')          .alias('sessionid_max')\n             ,f.countDistinct('session').alias('sessionid_unq')\n             ,f.count('session')        .alias('actions')\n             ,f.round(f.count('session')    /f.countDistinct('session'), 2).alias('avg_actions_by_session')\n             ,f.round(f.countDistinct('aid')/f.countDistinct('session'), 2).alias('unq_items_by_session')\n             ,f.round(f.avg(f.datediff('datetime_ref','datetime_min')) , 2).alias('weighted_avg_days_by_session')\n           ]\n\n    summary_stats = add_datetime_min(train)  .agg( f.lit('train').alias('dataset'), *agg_ops)\\\n             .union(add_datetime_min(valid)  .agg( f.lit('valid').alias('dataset'), *agg_ops))\\\n             .union(add_datetime_min(test_df).agg( f.lit('test' ).alias('dataset'), *agg_ops))\\\n             .toPandas()\n\n    display(summary_stats)\n    \n    # Extract pre-processed datasets as parquet\n    \n    ## Sampled datasets\n    TRAIN_SAMPLED_PATH = f\"/kaggle/working/train_sampled/FRACTION={FRACTION}/\"\n    if not os.path.isdir(TRAIN_SAMPLED_PATH):\n        print(f'Extracting data in path: `{TRAIN_SAMPLED_PATH}`')\n        train.write.mode(\"overwrite\").parquet(TRAIN_SAMPLED_PATH)\n    else:\n        print('Dataset has already been produced - not exporting.')\n        \n    TRAIN_LABELS_SAMPLED_PATH = f\"/kaggle/working/train_labels_sampled/FRACTION={FRACTION}/\"\n    if not os.path.isdir(TRAIN_LABELS_SAMPLED_PATH):\n        print(f'Extracting data in path: `{TRAIN_LABELS_SAMPLED_PATH}`')\n        train_labels.write.mode(\"overwrite\").parquet(TRAIN_LABELS_SAMPLED_PATH)\n    else:\n        print('Dataset has already been produced - not exporting.')\n        \n    VALID_SAMPLED_PATH = f\"/kaggle/working/valid_sampled/FRACTION={FRACTION}/\"\n    if not os.path.isdir(VALID_SAMPLED_PATH):\n        print(f'Extracting data in path: `{VALID_SAMPLED_PATH}`')\n        valid.write.mode(\"overwrite\").parquet(VALID_SAMPLED_PATH)\n    else:\n        print('Dataset has already been produced - not exporting.')\n        \n    VALID_LABELS_SAMPLED_PATH = f\"/kaggle/working/valid_labels_sampled/FRACTION={FRACTION}/\"\n    if not os.path.isdir(VALID_LABELS_SAMPLED_PATH):\n        print(f'Extracting data in path: `{VALID_LABELS_SAMPLED_PATH}`')\n        valid_labels.write.mode(\"overwrite\").parquet(VALID_LABELS_SAMPLED_PATH)\n    else:\n        print('Dataset has already been produced - not exporting.')","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:45:48.034054Z","iopub.execute_input":"2023-01-25T06:45:48.034430Z","iopub.status.idle":"2023-01-25T06:46:31.853762Z","shell.execute_reply.started":"2023-01-25T06:45:48.034400Z","shell.execute_reply":"2023-01-25T06:46:31.852516Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Delete checkpoint files\nif os.path.exists(v_checkPointPath):\n    shutil.rmtree(v_checkPointPath)","metadata":{"execution":{"iopub.status.busy":"2023-01-25T06:49:02.287066Z","iopub.execute_input":"2023-01-25T06:49:02.287473Z","iopub.status.idle":"2023-01-25T06:49:02.311202Z","shell.execute_reply.started":"2023-01-25T06:49:02.287441Z","shell.execute_reply":"2023-01-25T06:49:02.310243Z"},"trusted":true},"execution_count":null,"outputs":[]}]}