notebooks/experiment_local.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Criteo 1 TiB benchmark\n",
    "\n",
    "In this experiment we will evalutate a number of machine learning tools on a varying size of train data to determine how fast they learn, how much memory they consume etc.\n",
    "\n",
    "We will assess Vowpal Wabbit and XGBoost in local mode, and Spark.ML models in cluster mode.\n",
    "\n",
    "We will use terabyte click logs released by Criteo and sample needed amount of data from them.\n",
    "\n",
    "This instance of experiment notebook focuses on data preparation and training VW & XGBoost locally.\n",
    "\n",
    "Let's go!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Table of contents\n",
    "\n",
    "* [Configuration](#Configuration)\n",
    "* [Data preparation](#Data-preparation)\n",
    "  * [Criteo → LibSVM](#Criteo-→-LibSVM)\n",
    "  * [LibSVM → Train and test (sampling)](#LibSVM-→-Train-and-test-(sampling%29)\n",
    "  * [LibSVM train and test → VW train and test](#LibSVM-train-and-test-→-VW-train-and-test)\n",
    "  * [Local data](#Local-data)\n",
    "* [Local training](#Local-training)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%load_ext autotime\n",
    "%matplotlib inline\n",
    "\n",
    "from __future__ import print_function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Configuration\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Paths:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "criteo_data_remote_path = 'criteo/plain'\n",
    "libsvm_data_remote_path = 'criteo/libsvm'\n",
    "vw_data_remote_path = 'criteo/vw'\n",
    "\n",
    "local_data_path = 'criteo/data'\n",
    "local_results_path = 'criteo/results'\n",
    "local_runtime_path = 'criteo/runtime'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "\n",
    "criteo_day_template = os.path.join(criteo_data_remote_path, 'day_{}')\n",
    "libsvm_day_template = os.path.join(libsvm_data_remote_path, 'day_{}')\n",
    "vw_day_template = os.path.join(vw_data_remote_path, 'day_{}')\n",
    "\n",
    "libsvm_train_template = os.path.join(libsvm_data_remote_path, 'train', '{}')\n",
    "libsvm_test_template = os.path.join(libsvm_data_remote_path, 'test', '{}')\n",
    "vw_train_template = os.path.join(vw_data_remote_path, 'train', '{}')\n",
    "vw_test_template = os.path.join(vw_data_remote_path, 'test', '{}')\n",
    "\n",
    "local_libsvm_test_template = os.path.join(local_data_path, 'data.test.{}.libsvm')\n",
    "local_libsvm_train_template = os.path.join(local_data_path, 'data.train.{}.libsvm')\n",
    "local_vw_test_template = os.path.join(local_data_path, 'data.test.{}.vw')\n",
    "local_vw_train_template = os.path.join(local_data_path, 'data.train.{}.vw')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ensure_directory_exists(path):\n",
    "    if not os.path.exists(path):\n",
    "        os.makedirs(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Days to work on:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "days = list(range(0, 23 + 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Samples to take:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_samples = [\n",
    "    10000, 30000,  # tens of thousands\n",
    "    100000, 300000,  # hundreds of thousands\n",
    "    1000000, 3000000,  # millions\n",
    "    10000000, 30000000,  # tens of millions\n",
    "    100000000, 300000000,  # hundreds of millions\n",
    "    1000000000, 3000000000,  # billions\n",
    "]\n",
    "test_samples = [1000000]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark configuration and initialization:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total_cores = 256"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "executor_cores = 4\n",
    "executor_instances = total_cores / executor_cores\n",
    "memory_per_core = 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_name = 'Criteo experiment'\n",
    "\n",
    "master = 'yarn'\n",
    "\n",
    "settings = {\n",
    "    'spark.network.timeout': '600',\n",
    "    \n",
    "    'spark.driver.cores': '16',\n",
    "    'spark.driver.maxResultSize': '16G',\n",
    "    'spark.driver.memory': '32G',\n",
    "    \n",
    "    'spark.executor.cores': str(executor_cores),\n",
    "    'spark.executor.instances': str(executor_instances),\n",
    "    'spark.executor.memory': str(memory_per_core * executor_cores) + 'G',\n",
    "    \n",
    "    'spark.speculation': 'true',\n",
    "    'spark.yarn.queue': 'root.HungerGames',\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "\n",
    "builder = SparkSession.builder\n",
    "\n",
    "builder.appName(app_name)\n",
    "builder.master(master)\n",
    "for k, v in settings.items():\n",
    "    builder.config(k, v)\n",
    "\n",
    "spark = builder.getOrCreate()\n",
    "sc = spark.sparkContext\n",
    "\n",
    "sc.setLogLevel('ERROR')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Logging:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import logging\n",
    "reload(logging)\n",
    "\n",
    "\n",
    "handler = logging.StreamHandler(stream=sys.stdout)\n",
    "formatter = logging.Formatter('[%(asctime)s] %(message)s')\n",
    "handler.setFormatter(formatter)\n",
    "\n",
    "ensure_directory_exists(local_runtime_path)\n",
    "file_handler = logging.FileHandler(filename=os.path.join(local_runtime_path, 'mylog.log'), mode='a')\n",
    "file_handler.setFormatter(formatter)\n",
    "\n",
    "logger = logging.getLogger()\n",
    "logger.addHandler(handler)\n",
    "logger.addHandler(file_handler)\n",
    "logger.setLevel(logging.INFO)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logger.info('Spark version: %s.', spark.version)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data preparation\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Poor man's HDFS API:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def hdfs_exists(path):\n",
    "    l = !hadoop fs -ls $path 2>/dev/null\n",
    "    return len(l) != 0\n",
    "\n",
    "def hdfs_success(path):\n",
    "    return hdfs_exists(os.path.join(path, '_SUCCESS'))\n",
    "\n",
    "def hdfs_delete(path, recurse=False):\n",
    "    if recurse:\n",
    "        _ = !hadoop fs -rm -r $path\n",
    "    else:\n",
    "        _ = !hadoop fs -rm $path\n",
    "\n",
    "def hdfs_get(remote_path, local_path):\n",
    "    remote_path_glob = os.path.join(remote_path, 'part-*')\n",
    "    _ = !hadoop fs -cat $remote_path_glob >$local_path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load RDDs from one place and save them to another converted:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convert_chunked_data(input_path_template, output_path_template, chunks, load_rdd, convert_row, transform_rdd=None):\n",
    "    for chunk in chunks:\n",
    "        input_path = input_path_template.format(chunk)\n",
    "        output_path = output_path_template.format(chunk)\n",
    "\n",
    "        if hdfs_success(output_path):\n",
    "            logger.info('Chunk \"%s\" is already converted and saved to \"%s\", skipping.', chunk, output_path)\n",
    "            continue\n",
    "\n",
    "        logger.info('Reading chunk \"%s\" data from \"%s\".', chunk, input_path)\n",
    "        rdd = load_rdd(input_path)\n",
    "\n",
    "        if hdfs_exists(output_path):\n",
    "            logger.info('Cleaning \"%s\".', output_path)\n",
    "            hdfs_delete(output_path, recurse=True)\n",
    "\n",
    "        logger.info('Processing and saving to \"%s\".', output_path)\n",
    "        rdd = rdd.map(convert_row)\n",
    "        \n",
    "        if transform_rdd is not None:\n",
    "            rdd = transform_rdd(rdd)\n",
    "        \n",
    "        rdd.saveAsTextFile(output_path)\n",
    "\n",
    "        logger.info('Done with chunk \"%s\".', chunk)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Criteo → LibSVM\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Criteo RDD is actually a DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_criteo_rdd(path):\n",
    "    return (\n",
    "        spark\n",
    "        .read\n",
    "        .option('header', 'false')\n",
    "        .option('inferSchema', 'true')\n",
    "        .option('delimiter', '\\t')\n",
    "        .csv(path)\n",
    "        .rdd\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Simply add an index to each existing column except the first one which is a target:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def criteo_to_libsvm(row):\n",
    "    return (\n",
    "        str(row[0])\n",
    "        + ' '\n",
    "        + ' '.join(\n",
    "            [\n",
    "                # integer features\n",
    "                str(i) + ':' + str(row[i])\n",
    "                for i in range(1, 13 + 1)\n",
    "                if row[i] is not None\n",
    "            ] + [\n",
    "                # string features converted from hex to int\n",
    "                str(i) + ':' + str(int(row[i], 16))\n",
    "                for i in range(14, 39 + 1)\n",
    "                if row[i] is not None\n",
    "            ]\n",
    "        )\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Do it for all days:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "convert_chunked_data(criteo_day_template, libsvm_day_template, days, load_criteo_rdd, criteo_to_libsvm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### LibSVM → Train and test (sampling)\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's name samples as their shortened \"engineering\" notation - e.g. 1e5 is 100k etc.:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sample_name(sample):\n",
    "    return str(sample)[::-1].replace('000', 'k')[::-1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load data, sample a bit more than needed and cut at exact desired number of lines by zipping with index and filtering upto required index:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "oversample = 1.03\n",
    "sampled_partitions = 256\n",
    "\n",
    "\n",
    "def sample_and_save(input_path_template, output_path_template, days, samples):\n",
    "    union = None\n",
    "    union_count = None\n",
    "    \n",
    "    for sample in samples:\n",
    "        name = sample_name(sample)\n",
    "        output_path = output_path_template.format(name)\n",
    "        \n",
    "        if hdfs_success(output_path):\n",
    "            logger.info('Sample \"%s\" is already written to \"%s\", skipping.', sample, output_path)\n",
    "            continue\n",
    "            \n",
    "        logger.info('Preparing to write sample to \"%s\".', output_path)\n",
    "        \n",
    "        if union is None:\n",
    "            rdds = map(lambda day: sc.textFile(input_path_template.format(day)), days)\n",
    "            union = reduce(lambda left, right: left.union(right), rdds)\n",
    "\n",
    "            union_count = union.count()\n",
    "            logger.info('Total number of lines for days \"%s\" is \"%s\".', days, union_count)\n",
    "            \n",
    "        ratio = float(sample) / union_count\n",
    "        \n",
    "        sampled_union = (\n",
    "            union\n",
    "            .sample(False, min(1.0, oversample * ratio))\n",
    "            .zipWithIndex()\n",
    "            .filter(lambda z: z[1] < sample)\n",
    "            .map(lambda z: z[0])\n",
    "        )\n",
    "        \n",
    "        if hdfs_exists(output_path):\n",
    "            logger.info('Cleaning \"%s\".', output_path)\n",
    "            hdfs_delete(output_path, recurse=True)\n",
    "            \n",
    "        logger.info('Writing sample \"%s\" to \"%s\".', sample, output_path)\n",
    "        sampled_union.coalesce(sampled_partitions).saveAsTextFile(output_path)\n",
    "        \n",
    "        logger.info('Saved \"%s\" lines to \"%s\".', sc.textFile(output_path).count(), output_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sample all LibSVM data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_and_save(libsvm_day_template, libsvm_test_template, days[-1:], test_samples)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_and_save(libsvm_day_template, libsvm_train_template, days[:-1], train_samples)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### LibSVM train and test → VW train and test\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "LibSVM RDD is a text file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_libsvm_rdd(path):\n",
    "    return sc.textFile(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conversion is trivial - we only have to map target to {-1, 1} and convert categorical features to VW feature names as a whole:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def libsvm_to_vw(line):\n",
    "    parts = line.split(' ')\n",
    "    parts[0] = '1 |' if parts[0] == '1' else '-1 |'\n",
    "    for i in range(1, len(parts)):\n",
    "        index, _, value = parts[i].partition(':')\n",
    "        if int(index) >= 14:\n",
    "            parts[i] = index + '_' + value\n",
    "    return ' '.join(parts)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Also, data for VW should be well shuffled:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import hashlib\n",
    "\n",
    "\n",
    "def calculate_hash(something):\n",
    "    m = hashlib.md5()\n",
    "    m.update(str(something))\n",
    "    return m.hexdigest()\n",
    "\n",
    "def random_sort(rdd):\n",
    "    return (\n",
    "        rdd\n",
    "        .zipWithIndex()\n",
    "        .sortBy(lambda z: calculate_hash(z[1]))\n",
    "        .map(lambda z: z[0])\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert all LibSVM samples:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "convert_chunked_data(libsvm_test_template, vw_test_template, [sample_name(sample) for sample in test_samples], load_libsvm_rdd, libsvm_to_vw, transform_rdd=random_sort)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "convert_chunked_data(libsvm_train_template, vw_train_template, [sample_name(sample) for sample in train_samples], load_libsvm_rdd, libsvm_to_vw, transform_rdd=random_sort)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark is no longer needed:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Local data\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download all sampled data to local directory:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ensure_directory_exists(local_data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def count_lines(path):\n",
    "    with open(path) as f:\n",
    "        for i, _ in enumerate(f):\n",
    "            pass\n",
    "    return i + 1\n",
    "\n",
    "def download_data(remote_template, local_template, samples):\n",
    "    for sample in samples:\n",
    "        name = sample_name(sample)\n",
    "        remote_path = remote_template.format(name)\n",
    "        local_path = local_template.format(name)\n",
    "        if os.path.exists(local_path):\n",
    "            count = count_lines(local_path)\n",
    "            if count == sample:\n",
    "                logger.info('File \"%s\" is already loaded, skipping.', local_path)\n",
    "                continue\n",
    "            else:\n",
    "                logger.info('File \"%s\" already exists but number of lines \"%s\" is wrong (must be \"%s\"), reloading.', local_path, count, sample)\n",
    "        logger.info('Loading file \"%s\" as local file \"%s\".', remote_path, local_path)\n",
    "        hdfs_get(remote_path, local_path)\n",
    "        count = count_lines(local_path)\n",
    "        logger.info('File loaded to \"%s\", number of lines is \"%s\".', local_path, count)\n",
    "        assert count == sample, 'File \"{}\" contains wrong number of lines \"{}\" (must be \"{}\").'.format(local_path, count, sample)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "download_data(libsvm_test_template, local_libsvm_test_template, test_samples)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "download_data(libsvm_train_template, local_libsvm_train_template, train_samples)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "download_data(vw_test_template, local_vw_test_template, test_samples)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "download_data(vw_train_template, local_vw_train_template, train_samples)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Local training\n",
    "[_(back to toc)_](#Table-of-contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Measuring model quality and ML engine technical metrics:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys \n",
    "from matplotlib import pyplot\n",
    "from sklearn.metrics import (\n",
    "    auc,\n",
    "    log_loss,\n",
    "    roc_curve,\n",
    ")\n",
    "\n",
    "\n",
    "def measure(engine, sample, test_file, time_file, predictions_file):\n",
    "    \n",
    "    def get_last_in_line(s):\n",
    "        return s.rstrip().split( )[-1]\n",
    "\n",
    "    def parse_elapsed_time(s):\n",
    "        return reduce(lambda a, b: a * 60 + b, map(float, get_last_in_line(s).split(':')))\n",
    "\n",
    "    def parse_max_memory(s):\n",
    "        return int(get_last_in_line(s)) * 1024\n",
    "\n",
    "    def parse_cpu(s):\n",
    "        return float(get_last_in_line(s).rstrip('%')) / 100 \n",
    "\n",
    "\n",
    "    elapsed = -1\n",
    "    memory = -1\n",
    "    cpu = -1\n",
    "\n",
    "    with open(time_file, 'rb') as f:\n",
    "        for line in f:\n",
    "            if 'Elapsed (wall clock) time' in line:\n",
    "                elapsed = parse_elapsed_time(line)\n",
    "            elif 'Maximum resident set size' in line:\n",
    "                memory = parse_max_memory(line)\n",
    "            elif 'Percent of CPU' in line:\n",
    "                cpu = parse_cpu(line)\n",
    "\n",
    "    with open(test_file, 'rb') as f:\n",
    "        labels = [line.rstrip().split(' ')[0] == '1' for line in f]\n",
    "\n",
    "    with open(predictions_file, 'rb') as f:\n",
    "        scores = [float(line.rstrip().split(' ')[0]) for line in f]\n",
    "\n",
    "    fpr, tpr, _ = roc_curve(labels, scores)\n",
    "    roc_auc = auc(fpr, tpr)\n",
    "    ll = log_loss(labels, scores)\n",
    "    \n",
    "    figure = pyplot.figure(figsize=(6, 6))\n",
    "    pyplot.plot(fpr, tpr, linewidth=2.0)\n",
    "    pyplot.plot([0, 1], [0, 1], 'k--')\n",
    "    pyplot.xlabel('FPR')\n",
    "    pyplot.ylabel('TPR')\n",
    "    pyplot.title('{} {} - {:.3f} ROC AUC'.format(engine, sample_name(sample), roc_auc))\n",
    "    pyplot.show()\n",
    "\n",
    "    return {\n",
    "        'Engine': engine,\n",
    "        'Train size': sample,\n",
    "        'ROC AUC': roc_auc,\n",
    "        'Log loss': ll,\n",
    "        'Train time': elapsed,\n",
    "        'Maximum memory': memory,\n",
    "        'CPU load': cpu,\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Settings for VW & XGBoost and how to run them; I use (a little bit patched for correctness sake) GNU Time to measure running time, CPU load and memory consumption; configurations for VW & XGBoost are obtained via Hyperopt:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_time_command_and_file(train_file):\n",
    "    time_file = train_file + '.time'\n",
    "    return [\n",
    "        '/usr/local/bin/time',\n",
    "        '-v',\n",
    "        '--output=' + time_file,\n",
    "    ], time_file\n",
    "\n",
    "def get_vw_commands_and_predictions_file(train_file, test_file):\n",
    "    model_file = train_file + '.model'\n",
    "    predictions_file = test_file + '.predictions'\n",
    "    return [\n",
    "        'vw83',\n",
    "        '--link=logistic',\n",
    "        '--loss_function=logistic',\n",
    "        '-b', '29',\n",
    "        '-l', '0.3',\n",
    "        '--initial_t', '1',\n",
    "        '--decay_learning_rate', '0.5',\n",
    "        '--power_t', '0.5',\n",
    "        '--l1', '1e-15',\n",
    "        '--l2', '0',\n",
    "        '-d', train_file,\n",
    "        '-f', model_file,\n",
    "    ], [\n",
    "        'vw83',\n",
    "        '--loss_function=logistic',\n",
    "        '-t',\n",
    "        '-i', model_file,\n",
    "        '-d', test_file,\n",
    "        '-p', predictions_file,\n",
    "    ], predictions_file\n",
    "\n",
    "\n",
    "xgboost_conf = [\n",
    "    'booster = gbtree',\n",
    "    'objective = binary:logistic',\n",
    "    'nthread = 24',\n",
    "    'eval_metric = logloss',\n",
    "    'max_depth = 7',\n",
    "    'num_round = 200',\n",
    "    'eta = 0.2',\n",
    "    'gamma = 0.4',\n",
    "    'subsample = 0.8',\n",
    "    'colsample_bytree = 0.8',\n",
    "    'min_child_weight = 20',\n",
    "    'alpha = 3',\n",
    "    'lambda = 100',\n",
    "]\n",
    "\n",
    "\n",
    "def get_xgboost_commands_and_predictions_file(train_file, test_file, cache=False):\n",
    "    config_file = os.path.join(local_runtime_path, 'xgb.conf')\n",
    "    ensure_directory_exists(local_runtime_path)\n",
    "    with open(config_file, 'wb') as f:\n",
    "        for line in xgboost_conf:\n",
    "            print(line, file=f)\n",
    "    model_file = train_file + '.model'\n",
    "    predictions_file = test_file + '.predictions'\n",
    "    if cache:\n",
    "        train_file = train_file + '#' + train_file + '.cache'\n",
    "    return [\n",
    "        'xgboost',\n",
    "        config_file,\n",
    "        'data=' + train_file,\n",
    "        'model_out=' + model_file,\n",
    "    ], [\n",
    "        'xgboost',\n",
    "        config_file,\n",
    "        'task=pred',\n",
    "        'test:data=' + test_file,\n",
    "        'model_in=' + model_file,\n",
    "        'name_pred=' + predictions_file,\n",
    "    ], predictions_file\n",
    "\n",
    "def get_xgboost_ooc_commands_and_predictions_file(train_file, test_file):\n",
    "    return get_xgboost_commands_and_predictions_file(train_file, test_file, cache=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "engines = {\n",
    "    'vw': (get_vw_commands_and_predictions_file, local_vw_train_template, local_vw_test_template),\n",
    "    'xgb': (get_xgboost_commands_and_predictions_file, local_libsvm_train_template, local_libsvm_test_template),\n",
    "    'xgb.ooc': (get_xgboost_ooc_commands_and_predictions_file, local_libsvm_train_template, local_libsvm_test_template),\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Train & test everything:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "import subprocess\n",
    "\n",
    "\n",
    "measurements = []\n",
    "\n",
    "for sample in train_samples:\n",
    "    for engine in engines:\n",
    "        logger.info('Training \"%s\" on \"%s\" lines of data.', engine, sample)\n",
    "        \n",
    "        get_commands_and_predictions_file, train_template, test_template = engines[engine]\n",
    "\n",
    "        train_file = train_template.format(sample_name(sample))\n",
    "        test_file = test_template.format(sample_name(test_samples[0]))\n",
    "        logger.info('Will train on \"%s\" and test on \"%s\".', train_file, test_file)\n",
    "\n",
    "        command_time, time_file = get_time_command_and_file(train_file)\n",
    "        command_engine_train, command_engine_test, predictions_file = get_commands_and_predictions_file(train_file, test_file)\n",
    "\n",
    "        logger.info('Performing train.')\n",
    "        subprocess.call(command_time + command_engine_train)\n",
    "\n",
    "        logger.info('Performing test.')\n",
    "        subprocess.call(command_engine_test)\n",
    "\n",
    "        logger.info('Measuring results.')\n",
    "        measurement = measure(engine, sample, test_file, time_file, predictions_file)\n",
    "        logger.info(measurement)\n",
    "        measurements.append(measurement)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load measurements:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas\n",
    "\n",
    "\n",
    "measurements_df = pandas.DataFrame(measurements).sort_values(by=['Engine', 'Train size'])\n",
    "measurements_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Plot measurements:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "def extract_data_for_plotting(df, what):\n",
    "    return reduce(\n",
    "        lambda left, right: pandas.merge(left, right, how='outer', on='Train size'),\n",
    "        map(\n",
    "            lambda name: df[df.Engine == name][['Train size', what]].rename(columns={what: name}),\n",
    "            df.Engine.unique(),\n",
    "        ),\n",
    "    )   \n",
    "\n",
    "def plot_stuff(df, what, ylabel=None, **kwargs):\n",
    "    data = extract_data_for_plotting(df, what).set_index('Train size')\n",
    "    ax = data.plot(marker='o', figsize=(6, 6), title=what, grid=True, linewidth=2.0, **kwargs)  # xlim=(1e4, 4e9)\n",
    "    if ylabel is not None:\n",
    "        ax.set_ylabel(ylabel)\n",
    "\n",
    "\n",
    "plot_stuff(measurements_df, 'ROC AUC', logx=True)\n",
    "plot_stuff(measurements_df, 'Log loss', logx=True)\n",
    "plot_stuff(measurements_df, 'Train time', loglog=True, ylabel='s')\n",
    "plot_stuff(measurements_df, 'Maximum memory', loglog=True, ylabel='bytes')\n",
    "plot_stuff(measurements_df, 'CPU load', logx=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}