recfusion parallel exec.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f7ac4e2e-4b11-42ce-a1ff-a8f8ffa147b1",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# RecpackFusion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c19a803a-1f0e-4b4f-a3cb-4eca41b75cd9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "!pip install wandb einops &> /dev/null\n",
    "!rm -r recpackfusion\n",
    "!git clone https://github.com/gabriben/recfusion.git\n",
    "!cd recfusion; pip install . &> /dev/null"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b694f50b-50be-434c-916d-e81690d7b465",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from recpack.quick import quick_train\n",
    "\n",
    "\n",
    "model = 'RecFusionMLP'\n",
    "dataset = 'MovieLens1M'\n",
    "prep_hypers = {\n",
    "    'ds_path': '/dbfs/datasets/RecSys/',  \n",
    "    'min_rating': 4,\n",
    "    'min_items_per_user': 5,\n",
    "    'min_users_per_item': 5,\n",
    "    'generalization': 'StrongGeneralization', # WeakGeneralization,\n",
    "    \"train_val_test\": [0.8, 0.1, 0.1],\n",
    "    'force_even_items': False\n",
    "}\n",
    "train_hypers = {\n",
    "    # \"max_epochs\":1,\n",
    "    \"stop_early\": False, # \"max_iter_no_change\" : 10, \"min_improvement\": 0.001,\n",
    "    # \"validation_sample_size\": 1000,\n",
    "    # \"dim_bottleneck_layer\" : 128,\n",
    "    'T': 2, \n",
    "    # 'p_dnns_depth': 5,\n",
    "    # \"T\": 100,\n",
    "    # \"batch_size\" : 200,\n",
    "    # \"M\" : 200,\n",
    "    \"p_dnns_depth\": 2,\n",
    "    # \"anneal_steps\" : 20    \n",
    "    # \"time_embedding_as_input\" : True  \n",
    "    #\"schedule_type\": 'fixed',\n",
    "    #'jascha_bin_process': True,\n",
    "    #'b_start' : 0.1,    \n",
    "}\n",
    "val_metric = {'NDCGK':100} #{'NDCGK':100}    \n",
    "test_metrics = {\n",
    "    'NDCGK' : [10, 20, 50, 100],\n",
    "    'RecallK' : [10, 20, 50],\n",
    "    'HitK': [20, 50, 100],\n",
    "    'CalibratedRecallK': [10, 20, 50]\n",
    "} "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "901f1664-9be1-4cdb-956e-8924e91abe49",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "neural_baselines = [\"RecVAE\", \"CODIGEM\", \"MultVAE\"] \n",
    "recfusion = [\"RecFusionBin\", \"RecFusionMLPVar\", \"RecFusionMLPT\", \"RecFusionMLP\"]\n",
    "recfusion_unet = [\"RecFusionUnet1D\", \"RecFusionUnet2D\"]\n",
    "basic_baselines = [\"Random\", \"Popularity\", \"EASE\", \"SLIM\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "316d118a-0cbe-48eb-ad00-269d93b3ac7c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from pyspark.sql.functions import col, udf\n",
    "from pyspark.sql.types import LongType\n",
    "import os\n",
    "\n",
    "# Declare the function and create the UDF\n",
    "@udf(\"float\")\n",
    "def training_func(model):\n",
    "\n",
    "  os.environ[\"WANDB_API_KEY\"] = \"\"\n",
    "  os.environ[\"WANDB_PROJECT\"] = \"\"\n",
    "\n",
    "  m, v = quick_train(model, dataset, prep_hypers, train_hypers, val_metric, test_metrics)\n",
    "  loss = v\n",
    "  return v\n",
    "\n",
    "# The df contains on each row a combination of the hyper parameters\n",
    "df = spark.createDataFrame(pd.DataFrame({\n",
    "    \"model\": [\"RecFusionMLP\"] * 10\n",
    "    # \"model\": # basic_baselines\n",
    "}))\n",
    "\n",
    "df.repartition(df.count()).withColumn(\"loss\", training_func(col(\"model\"))).show()"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 1588776262876761,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 2
   },
   "notebookName": "recfusion parallel exec clean",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}