notebooks/Parse Json to Pandas Dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parse Json to Pandas Dataframes\n",
    "This script is used to convert json packets into a dictionary where the key is a unique metadata configuration and the value is a Pandas dataframe. The Pandas dataframe has a ds column and a y column corresponding to the timestamp and corresponding value in the time series. The dictionary is then stored in a Pickle file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import pandas as pd\n",
    "import fnmatch\n",
    "import os\n",
    "import bz2\n",
    "import pickle\n",
    "import gc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# read files in list and convert to pandas dataframes\n",
    "def load_files(files, file_format):\n",
    "    dfs = {}\n",
    "    for file in files:\n",
    "        # check file format and read appropriately\n",
    "        if file_format == \".json\":\n",
    "            f = open(file, 'rb')\n",
    "        else:\n",
    "            f = bz2.BZ2File(file, 'rb')\n",
    "        jsons = json.load(f)\n",
    "        f.close()\n",
    "\n",
    "        # iterate through packets in file\n",
    "        for pkt in jsons:\n",
    "            # create a new dataframe with packet timestamp and values\n",
    "            df = pd.DataFrame.from_dict(pkt[\"values\"])\n",
    "            df = df.rename( columns={0:\"ds\", 1:\"y\"})\n",
    "            df[\"ds\"] = pd.to_datetime(df[\"ds\"], unit='s')\n",
    "            df = df.sort_values(by=[\"ds\"])\n",
    "            df.y = pd.to_numeric(df['y'], errors='coerce')\n",
    "            df = df.dropna()\n",
    "            md = str(pkt[\"metric\"])\n",
    "            # append generated dataframe and metadata to collection\n",
    "            try:\n",
    "                dfs[md] = dfs[md].append(df, ignore_index=True)\n",
    "            except:\n",
    "                dfs[md] = df\n",
    "    return dfs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# take a list of dataframes and their metadata and collapse to a\n",
    "# collection of unique time series (based on unique metadata)\n",
    "def collapse_to_unique(dfs_master, dfs_new):\n",
    "    # iterate through metadata\n",
    "    dfs_remaining = {}\n",
    "    for md in dfs_new.keys():\n",
    "        try:\n",
    "            # find metadata in our master list\n",
    "            # if this throws an error, simply add it to the list\n",
    "            dfs_master[md] = dfs_master[md].append(dfs_new[md], ignore_index=True)\n",
    "        except:\n",
    "            dfs_remaining[md] = dfs_new[md]\n",
    "    return dfs_master, dfs_remaining"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create pickle file containing data\n",
    "def save_checkpoint(pds, file):\n",
    "    if file[-4:] != \".pkl\":\n",
    "        file = file + \".pkl\"\n",
    "    f = open(file, \"wb\")\n",
    "    pickle.dump(pds, f)\n",
    "    f.close()\n",
    "    return file\n",
    "\n",
    "# load pickle file containing data\n",
    "def load_checkpoint(file):\n",
    "    f = open(file, \"rb\")\n",
    "    pds = pickle.load(f)\n",
    "    f.close()\n",
    "    return pds\n",
    "# remove all temp pickle files generated during this program\n",
    "def combine_checkpoints(master_file):\n",
    "    df = {}\n",
    "    files = os.listdir()\n",
    "    for file in files:\n",
    "        if fnmatch.fnmatch(file, \"collapsed_*.pkl\"):\n",
    "            try:\n",
    "                f = open(file, \"rb\")\n",
    "                dfs = pickle.load(f)\n",
    "                f.close()\n",
    "                df.update(dfs)\n",
    "            except:\n",
    "                continue\n",
    "            os.system(\"rm \" + file)\n",
    "        elif fnmatch.fnmatch(file, \"raw_*.pkl\"):\n",
    "            os.system(\"rm \" + file)\n",
    "    f = open(master_file + \".pkl\", \"wb\")\n",
    "    pickle.dump(df, f)\n",
    "    f.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load all files and convert to a list of pandas dataframes\n",
    "def convert_to_pandas(files, file_format, batch_size):\n",
    "    checkpoints = []\n",
    "    # # separate files into batches\n",
    "    batches = [files[batch_size*i:batch_size*(i+1)] for i in range(int(len(files)/batch_size) + 1)]\n",
    "    print(\"num_batches\", len(batches))\n",
    "    i = 0\n",
    "    for batch in batches:\n",
    "        print(i)\n",
    "        i += 1\n",
    "        # get new portion of dataframes and add to master set\n",
    "        pds_new = load_files(batch, file_format)\n",
    "        cp = save_checkpoint(pds_new, \"raw_\" + str(i))\n",
    "        checkpoints.append(cp)\n",
    "        gc.collect()\n",
    "\n",
    "    pds = []\n",
    "    # iterate checkpoint by checkpoint and add data to unique collection\n",
    "    # of time series\n",
    "    collapsed_fs = []\n",
    "    i = 0\n",
    "    for cp in checkpoints:\n",
    "        i += 1\n",
    "        print(i)\n",
    "        pds_new = load_checkpoint(cp)\n",
    "        print(i)\n",
    "        # load data in batches and combine dataframes\n",
    "        for f in collapsed_fs:\n",
    "            pds = load_checkpoint(f)\n",
    "            pds, pds_new = collapse_to_unique(pds, pds_new)\n",
    "            save_checkpoint(pds, f)\n",
    "            gc.collect()\n",
    "        if len(pds_new) > 0:\n",
    "            f_new = save_checkpoint(pds_new, \"collapsed_\" + str(i)) \n",
    "            print(\"Generated \", f_new)\n",
    "            collapsed_fs.append(f_new)   \n",
    "        print(i)\n",
    "        gc.collect()\n",
    "    return pds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get all appropriately formatted files in a folder\n",
    "def retrieve_filenames(path, file_format):\n",
    "    filenames = []\n",
    "    for file in os.listdir(path):\n",
    "            # check if this file has correct ending (regex)\n",
    "            if fnmatch.fnmatch(file, \"*\" + file_format):\n",
    "                f_name = path + file\n",
    "                filenames.append(f_name)\n",
    "    return filenames\n",
    "\n",
    "# get main input arguments and return formatted data\n",
    "def read_input(data_folder, metric, file_format, batch_size):\n",
    "    # metric-specific data folder\n",
    "    folder = data_folder + metric + \"/\"\n",
    "    # get all files in folder\n",
    "    files = os.listdir(folder)\n",
    "\n",
    "    # automatically detect metric type\n",
    "    if \"quantile\" in files:\n",
    "        metric_type = \"summary\"\n",
    "        label = \"quantile\"\n",
    "        filenames = retrieve_filenames(folder + \"quantile/\", file_format)\n",
    "#         filenames_count = retrieve_filenames(folder + \"count/\", file_format)\n",
    "#         filenames_sum = retrieve_filenames(folder + \"sum/\", file_format)\n",
    "    elif \"bucket\" in files:\n",
    "        metric_type = \"histogram\"\n",
    "        label = \"le\"\n",
    "        filenames = retrieve_filenames(folder + \"bucket/\", file_format)\n",
    "#         filenames_count = retrieve_filenames(folder + \"count/\", file_format)\n",
    "#         filenames_sum = retrieve_filenames(folder + \"sum/\", file_format)\n",
    "    else:\n",
    "        metric_type = \"counter/gauge\"\n",
    "        label = \"\"\n",
    "        filenames = retrieve_filenames(folder, file_format)\n",
    "    \n",
    "    pd_frames = convert_to_pandas(filenames, file_format, batch_size)\n",
    "\n",
    "    return pd_frames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def main():\n",
    "    print(\"Formatting Data\")\n",
    "    pd_frames = read_input(input_dir, metric, fformat, batch_size)\n",
    "    print(\"Conversion successful\")\n",
    "\n",
    "    master_file = output_dir + metric\n",
    "\n",
    "    combine_checkpoints(master_file)\n",
    "\n",
    "    print(\"Saved data:\", master_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Formatting Data\n",
      "num_batches 11\n",
      "0\n",
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "5\n",
      "6\n",
      "7\n",
      "8\n",
      "9\n",
      "10\n",
      "1\n",
      "1\n",
      "Generated  collapsed_1.pkl\n",
      "1\n",
      "2\n",
      "2\n",
      "Generated  collapsed_2.pkl\n",
      "2\n",
      "3\n",
      "3\n",
      "Generated  collapsed_3.pkl\n",
      "3\n",
      "4\n",
      "4\n",
      "Generated  collapsed_4.pkl\n",
      "4\n",
      "5\n",
      "5\n",
      "Generated  collapsed_5.pkl\n",
      "5\n",
      "6\n",
      "6\n",
      "Generated  collapsed_6.pkl\n",
      "6\n",
      "7\n",
      "7\n",
      "Generated  collapsed_7.pkl\n",
      "7\n",
      "8\n",
      "8\n",
      "Generated  collapsed_8.pkl\n",
      "8\n",
      "9\n",
      "9\n",
      "Generated  collapsed_9.pkl\n",
      "9\n",
      "10\n",
      "10\n",
      "Generated  collapsed_10.pkl\n",
      "10\n",
      "11\n",
      "11\n",
      "11\n",
      "Conversion successful\n",
      "Saved data: http_request_duration_microseconds\n"
     ]
    }
   ],
   "source": [
    "if __name__ == '__main__':\n",
    "\n",
    "    # input parameters\n",
    "    metric = \"http_request_duration_microseconds\"\n",
    "    fformat='.json.bz2'\n",
    "    input_dir = \"data/\"\n",
    "    output_dir = \"\"\n",
    "    batch_size= 20\n",
    "\n",
    "    main()"
   ]
  }
 ],
 "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}