correlation_single_rack_one_day.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from distutils.dir_util import copy_tree\n",
    "import itertools\n",
    "import scipy\n",
    "from scipy import stats\n",
    "import dask.dataframe as dd\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import random\n",
    "from pyspark.sql import SparkSession\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generic config\n",
    "\n",
    "# The location of the dataset, with all the sub-folders (one folder per metric)\n",
    "root_dataset = \"/path/to/processed-surf-dataset/\"\n",
    "store_location = \"./SurfCoefficients50days/\"\n",
    "\n",
    "# Create the store_location if not exists\n",
    "os.makedirs(store_location, exist_ok=True)\n",
    "\n",
    "cores = os.cpu_count()\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[{}]\".format(cores)) \\\n",
    "    .appName(\"Test\") \\\n",
    "    .config(\"spark.executor.memory\", \"30G\") \\\n",
    "    .config(\"spark.driver.memory\", \"30G\") \\\n",
    "    .config(\"spark.driver.maxResultSize\", \"30G\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "def correlate_two_dfs(df, df2, coef_file):\n",
    "    import logging\n",
    "    logger = logging.getLogger(__name__)\n",
    "    try:\n",
    "        # merge them on index and node\n",
    "        merged = df.merge(df2, on=['index', 'node'], copy=False, how='inner')\n",
    "        del df2  # Reduce RAM usage\n",
    "        merged.drop(['index', 'node'], axis=1, inplace=True)\n",
    "        merged.dropna(inplace=True) # Drop rows with nan in df1 or df2\n",
    "        \n",
    "        # Get 1k samples to speed up the computation significantly, following https://link.springer.com/article/10.1007/BF02294183 this should be more than sufficient.\n",
    "        \n",
    "        # Using the pandas dataframe, compute the Pearson, Spearman, and Kendall correlations + pvalues\n",
    "        values = [scipy.stats.pearsonr(merged['df1'], merged['df2']), \n",
    "                  scipy.stats.spearmanr(merged['df1'], merged['df2']), \n",
    "                  scipy.stats.kendalltau(merged['df1'], merged['df2']),\n",
    "                 ]\n",
    "\n",
    "        correlations = [val[0] for val in values]\n",
    "        pvalues = [val[1] for val in values]\n",
    "        \n",
    "        if any([abs(val) >= 0.8 for val in correlations]):\n",
    "            print(coef_file)\n",
    "\n",
    "        with open(coef_file, \"w\") as file1:\n",
    "            file1.write(\"|\".join([str(x) for x in correlations]))\n",
    "            file1.write(\"\\n\")\n",
    "            file1.write(\"|\".join([str(x) for x in pvalues]))\n",
    "\n",
    "    except Exception:\n",
    "        logger.exception(\"Fatal error while computing correlations\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get all folder names in the root_dataset\n",
    "folders = next(os.walk(root_dataset))[1]\n",
    "random.shuffle(folders)  # Shuffle the folders so that if we run another node in parallel, the chance on collisions is very small\n",
    "\n",
    "for i in range(len(folders)):\n",
    "    # Make sure you run this with pyarrow 1.0 or higher, 0.17 gives an error for directories which was fixed in 1.0!\n",
    "    if not os.path.isfile(os.path.join(root_dataset, folders[i], \"1579474800_1579561185.parquet\")): continue\n",
    "    df = pd.read_parquet(os.path.join(root_dataset, folders[i], \"1579474800_1579561185.parquet\"))\n",
    "    if isinstance(df.columns, pd.MultiIndex):  # For GPU nodes that have a multi-index, average the value of the cards\n",
    "        df = df.groupby(axis=1, level=0).mean()\n",
    "\n",
    "    # Make the index a date index\n",
    "    df.index = pd.to_datetime(df.index, unit=\"s\")\n",
    "\n",
    "    # Make sure to replace invalid values with nan so we can filter them later.\n",
    "    df.replace(-1, np.nan, inplace=True)\n",
    "\n",
    "    # Convert the dataframe to long form using melt\n",
    "    df.reset_index(inplace=True)\n",
    "    df = df.melt(id_vars=['index'], var_name=\"node\", value_name=\"df1\")\n",
    "\n",
    "    # To enable comparisons, convert values to doubles\n",
    "    # Not sure if this is needed - Exception: ArrowTypeError('fields had matching names but differing types. From: r1123n7: int64 To: r1123n7: double')\n",
    "    df = df.astype({\"df1\": np.double}, copy=False)\n",
    "    \n",
    "    for j in range(i+1, len(folders)):\n",
    "        coef_file = os.path.join(store_location, \"{}+{}_correlations.csv\".format(folders[i], folders[j]))\n",
    "    \n",
    "        \n",
    "        if os.path.isfile(coef_file): continue\n",
    "        if not os.path.isfile(os.path.join(root_dataset, folders[j], \"1579474800_1579561185.parquet\")): continue\n",
    "\n",
    "        # Same for dir 2\n",
    "        df2 = pd.read_parquet(os.path.join(root_dataset, folders[j], \"1579474800_1579561185.parquet\"))\n",
    "        if isinstance(df2.columns, pd.MultiIndex):  # For GPU nodes that have a multi-index, average the value of the cards\n",
    "            df2 = df2.groupby(axis=1, level=0).mean()\n",
    "\n",
    "        df2.index = pd.to_datetime(df2.index, unit=\"s\")\n",
    "        df2.replace(-1, np.nan, inplace=True)\n",
    "        df2.reset_index(inplace=True)\n",
    "        df2 = df2.melt(id_vars=['index'], var_name=\"node\", value_name=\"df2\")\n",
    "        df2 = df2.astype({\"df2\": np.double}, copy=False)\n",
    "        \n",
    "        correlate_two_dfs(df, df2, coef_file)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}