ExampleCountingStreamsMisraGries.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "7e79f2f6",
   "metadata": {},
   "source": [
    "# Example: Counting Streams: MisraGries Algorithm\n",
    "\n",
    "The <b>Misra-Gries algorithm</b> identifies items that appear <i>more than</i>\n",
    "<b>L/N</b> times in in a stream of length <b>L</b>, where <b>N</b> is a \n",
    "parameter of the algorithm. Such items are called <b><i>heavy hitters</i></b>\n",
    "of the stream. For a stream of a billion elements, and with <b>N</b> = 1000, \n",
    "a heavy hitter is an item that appears more than a million times in the stream.\n",
    "\n",
    "The algorithm operates in two phases: in the first phase it identifies candidates\n",
    "for heavy hitters and in the next phase it determines whether these candidates\n",
    "are indeed heavy hitters by running the candidates across the entire stream.\n",
    "The first phase guarantees that every heavy hitter is a candidate; however, it\n",
    "does not guarantee that every candidate is a heavy hitter. For some applications,\n",
    "the first phase is sufficient.\n",
    "   \n",
    "This code implements the first phase of the algorithm.\n",
    "This code has data structures and print statements purely to help to teach the\n",
    "algorithm; these statements should be removed in an application.\n",
    "\n",
    "See:\n",
    "    http://www.cs.utexas.edu/users/misra/Notes.dir/HeavyHitters.pdf."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3277ae6e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def misra_gries_process_element(v, candidates, inputs, N):\n",
    "    \"\"\"\n",
    "    Parameters\n",
    "    ----------\n",
    "    v: object\n",
    "        An element of the input stream\n",
    "    candidates: dict\n",
    "        key: item of input stream\n",
    "        value: int\n",
    "             A lower bound on the number of times the key has\n",
    "             appeared on the input stream.\n",
    "        initially: empty\n",
    "    inputs: dict\n",
    "        key: item\n",
    "        value: number of times the item appears in the stream\n",
    "        initially: empty\n",
    "        THIS IS USED ONLY FOR DEBUGGING AND EXPLANATION!\n",
    "        REMOVE FOR USE IN AN APPLICATION.\n",
    "    N: positive integer (constant)\n",
    "       A heavy hitter appears more than L/N times in a stream\n",
    "       of length L.\n",
    "    \n",
    "\n",
    "    Returns: None\n",
    "    -------\n",
    "       Updates candidates and inputs\n",
    "\n",
    "    \"\"\"\n",
    "\n",
    "    # Step 1. If the input element, v, is already a candidate then\n",
    "    # increment v' count.\n",
    "    if v in candidates: candidates[v] += 1\n",
    "    # Step 2. If v is not already a candidate, and there are fewer\n",
    "    # than N-1 candidates, then add v as a candidate.\n",
    "    elif len(candidates) < N-1: candidates[v] = 1\n",
    "    # Step 3. If v is not already a candidate and there are already\n",
    "    # N-1 candidates then decrement the count for all candidates\n",
    "    else:\n",
    "        for key, value in candidates.items():\n",
    "            candidates[key] -= 1\n",
    "    #Step 4. Remove candidates whose count is reduced to 0.\n",
    "    zero_count_candidates = \\\n",
    "      [key for key in candidates.keys() if candidates[key] == 0]\n",
    "    for candidate in zero_count_candidates:\n",
    "        del candidates[candidate]\n",
    "                \n",
    "                \n",
    "    # FOR DEBUGGING AND EXPLANATION ONLY: UPDATE INPUTS\n",
    "    if v in inputs.keys(): inputs[v] += 1\n",
    "    else: inputs[v] = 1\n",
    "\n",
    "    # PRINT ONLY FOR EXPLANATION / TEACHING\n",
    "    print('inputs')\n",
    "    print(inputs)\n",
    "    print('candidates')\n",
    "    print(candidates)\n",
    "    print('remove non heavy hitters from candidates')\n",
    "    sum_values = sum([value for value in candidates.values()])\n",
    "    smaller_candidate_set = \\\n",
    "      [key for key in candidates.keys()\n",
    "           if candidates[key] >=sum_values/N]\n",
    "    print(smaller_candidate_set)\n",
    "    print()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "a8f0be4a",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_Misra_Gries():\n",
    "    from stream import Stream, run\n",
    "    from example_operators import single_item\n",
    "    \n",
    "    # Declare streams.\n",
    "    x = Stream('input')\n",
    "    # Create the Agent\n",
    "    single_item(in_stream=x,\n",
    "        func=misra_gries_process_element, candidates={},\n",
    "        inputs={}, N=3)\n",
    "\n",
    "    # Put data into streams and run.\n",
    "    x.extend([3])\n",
    "    run()\n",
    "\n",
    "    x.extend([2])\n",
    "    run()\n",
    "\n",
    "\n",
    "    x.extend([1])\n",
    "    run()\n",
    "\n",
    "    x.extend([3, 3, 3])\n",
    "\n",
    "    x.extend([2, 2])\n",
    "    run()\n",
    "\n",
    "    x.extend([4, 4, 4, 4, 4, 4,])\n",
    "    run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d92bb924",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "inputs\n",
      "{3: 1}\n",
      "candidates\n",
      "{3: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 1, 2: 1}\n",
      "candidates\n",
      "{3: 1, 2: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3, 2]\n",
      "\n",
      "inputs\n",
      "{3: 1, 2: 1, 1: 1}\n",
      "candidates\n",
      "{}\n",
      "remove non heavy hitters from candidates\n",
      "[]\n",
      "\n",
      "inputs\n",
      "{3: 2, 2: 1, 1: 1}\n",
      "candidates\n",
      "{3: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 3, 2: 1, 1: 1}\n",
      "candidates\n",
      "{3: 2}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 1, 1: 1}\n",
      "candidates\n",
      "{3: 3}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 2, 1: 1}\n",
      "candidates\n",
      "{3: 3, 2: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1}\n",
      "candidates\n",
      "{3: 3, 2: 2}\n",
      "remove non heavy hitters from candidates\n",
      "[3, 2]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 1}\n",
      "candidates\n",
      "{3: 2, 2: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3, 2]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 2}\n",
      "candidates\n",
      "{3: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 3}\n",
      "candidates\n",
      "{3: 1, 4: 1}\n",
      "remove non heavy hitters from candidates\n",
      "[3, 4]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 4}\n",
      "candidates\n",
      "{3: 1, 4: 2}\n",
      "remove non heavy hitters from candidates\n",
      "[3, 4]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 5}\n",
      "candidates\n",
      "{3: 1, 4: 3}\n",
      "remove non heavy hitters from candidates\n",
      "[4]\n",
      "\n",
      "inputs\n",
      "{3: 4, 2: 3, 1: 1, 4: 6}\n",
      "candidates\n",
      "{3: 1, 4: 4}\n",
      "remove non heavy hitters from candidates\n",
      "[4]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_Misra_Gries()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}