{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Phi_K spark tutorial\n",
    "\n",
    "This notebook shows you how to obtain the Phi_K correlation matrix for a spark dataframe.\n",
    "Calculating the Phi_K matrix consists of two steps:\n",
    "\n",
    "- Obtain the 2d contingency tables for all variable pairs. To make these we use the [`histogrammar` package](https://github.com/histogrammar/histogrammar-python).\n",
    "- Calculate the Phi_K value for each variable pair from its contingency table.\n",
    "\n",
    "Make sure you install the histogrammar package to make the 2d histograms, that are then used to calculate phik."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%capture\n",
    "# install histogrammar (if not installed yet)\n",
    "import sys\n",
    "\n",
    "!\"{sys.executable}\" -m pip install histogrammar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import itertools\n",
    "\n",
    "import pandas as pd\n",
    "import histogrammar as hg\n",
    "from histogrammar.plot.hist_numpy import get_2dgrid\n",
    "\n",
    "import phik\n",
    "from phik import resources\n",
    "from phik.phik import spark_phik_matrix_from_hist2d_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# histogramming is done using the histogrammar library"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import __version__ as pyspark_version\n",
    "\n",
    "scala = '2.12' if int(pyspark_version[0]) >= 3 else '2.11'\n",
    "hist_jar = f'io.github.histogrammar:histogrammar_{scala}:1.0.20'\n",
    "hist_spark_jar = f'io.github.histogrammar:histogrammar-sparksql_{scala}:1.0.20'\n",
    "\n",
    "spark = SparkSession.builder.config(\n",
    "    \"spark.jars.packages\", f'{hist_spark_jar},{hist_jar}'\n",
    ").getOrCreate()\n",
    "\n",
    "spark = SparkSession.builder.config(\n",
    "    \"spark.jars.packages\", f'{hist_spark_jar},{hist_jar}'\n",
    ").getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data\n",
    "\n",
    "A simulated dataset is part of the phik-package. The dataset concerns fake car insurance data. Load the dataset here:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = pd.read_csv( resources.fixture('fake_insurance_data.csv.gz') )\n",
    "sdf = spark.createDataFrame(data)\n",
    "sdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "combis = itertools.combinations_with_replacement(sdf.columns, 2)\n",
    "combis = [list(c) for c in combis]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(combis)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# step 1: create histograms (this runs spark histogrammar in the background)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# see the doc-string of hg_make_histograms() for binning options.\n",
    "hists = sdf.hg_make_histograms(combis)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# collect the numpy contingency tables into a dict\n",
    "grids = {k:(get_2dgrid(h)[2]) for k,h in hists.items()}\n",
    "print(grids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# we can store the histograms if we want to\n",
    "if False:\n",
    "    import pickle\n",
    "\n",
    "    with open('grids.pkl', 'wb') as outfile:\n",
    "        pickle.dump(grids, outfile)\n",
    "\n",
    "    with open('grids.pkl', 'rb') as handle:\n",
    "        grids = pickle.load(handle)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# step 2: calculate phik matrix (runs rdd parallellization over all 2d histograms)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "phik_matrix = spark_phik_matrix_from_hist2d_dict(sc, grids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "phik_matrix"
   ]
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}