beam-dev-env/section2/minute_traffic.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "8d80de74-cf38-402f-90d3-e0d190a27863",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "import typing\n",
    "import datetime\n",
    "\n",
    "import apache_beam as beam\n",
    "import apache_beam.runners.interactive.interactive_beam as ib\n",
    "from apache_beam.runners.interactive import interactive_runner\n",
    "from apache_beam.transforms.combiners import CountCombineFn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "d6d54a3b-8dbe-42c3-a86a-c386f61dd0eb",
   "metadata": {},
   "outputs": [],
   "source": [
    "class EventLog(typing.NamedTuple):\n",
    "    ip: str\n",
    "    id: str\n",
    "    lat: float\n",
    "    lng: float\n",
    "    user_agent: str\n",
    "    age_bracket: str\n",
    "    opted_into_marketing: bool\n",
    "    http_request: str\n",
    "    http_response: int\n",
    "    file_size_bytes: int\n",
    "    event_datetime: str\n",
    "    event_ts: int\n",
    "\n",
    "beam.coders.registry.register_coder(EventLog, beam.coders.RowCoder)\n",
    "\n",
    "def parse_json(element: str):\n",
    "    row = json.loads(element)\n",
    "    # lat/lng sometimes empty string\n",
    "    if not row[\"lat\"] or not row[\"lng\"]:\n",
    "        row = {**row, **{\"lat\": -1, \"lng\": -1}}\n",
    "    return EventLog(**row)\n",
    "\n",
    "def add_timestamp(element: EventLog):\n",
    "    ts = datetime.datetime.strptime(element.event_datetime, \"%Y-%m-%dT%H:%M:%S.%f\").timestamp()\n",
    "    return beam.window.TimestampedValue(element, ts)\n",
    "\n",
    "class AddWindowTS(beam.DoFn):\n",
    "    def process(self, element: int, window=beam.DoFn.WindowParam):\n",
    "        window_start = window.start.to_utc_datetime().isoformat(timespec=\"seconds\")\n",
    "        window_end = window.end.to_utc_datetime().isoformat(timespec=\"seconds\")\n",
    "        output = {\"window_start\": window_start, \"window_end\": window_end, \"page_views\": element}\n",
    "        yield output\n",
    "\n",
    "PARENT_DIR = os.path.dirname(os.getcwd())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "cf04efc3-e5e9-43f8-b234-ea851466f42a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "\n",
       "        if (typeof window.interactive_beam_jquery == 'undefined') {\n",
       "          var jqueryScript = document.createElement('script');\n",
       "          jqueryScript.src = 'https://code.jquery.com/jquery-3.4.1.slim.min.js';\n",
       "          jqueryScript.type = 'text/javascript';\n",
       "          jqueryScript.onload = function() {\n",
       "            var datatableScript = document.createElement('script');\n",
       "            datatableScript.src = 'https://cdn.datatables.net/1.10.20/js/jquery.dataTables.min.js';\n",
       "            datatableScript.type = 'text/javascript';\n",
       "            datatableScript.onload = function() {\n",
       "              window.interactive_beam_jquery = jQuery.noConflict(true);\n",
       "              window.interactive_beam_jquery(document).ready(function($){\n",
       "                \n",
       "              });\n",
       "            }\n",
       "            document.head.appendChild(datatableScript);\n",
       "          };\n",
       "          document.head.appendChild(jqueryScript);\n",
       "        } else {\n",
       "          window.interactive_beam_jquery(document).ready(function($){\n",
       "            \n",
       "          });\n",
       "        }"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "\n",
       "            <link rel=\"stylesheet\" href=\"https://stackpath.bootstrapcdn.com/bootstrap/4.4.1/css/bootstrap.min.css\" integrity=\"sha384-Vkoo8x4CGsO3+Hhxv8T/Q5PaXtkKtu6ug5TOeNV6gBiFeWPGFN9MuhOf23Q9Ifjh\" crossorigin=\"anonymous\">\n",
       "            <div id=\"progress_indicator_edfb8bc471267ecdc4353329135e82e5\">\n",
       "              <div class=\"spinner-border text-info\" role=\"status\"></div>\n",
       "              <span class=\"text-info\">Processing... show</span>\n",
       "            </div>\n",
       "            "
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "\n",
       "            <style>\n",
       "            .p-Widget.jp-OutputPrompt.jp-OutputArea-prompt:empty {\n",
       "              padding: 0;\n",
       "              border: 0;\n",
       "            }\n",
       "            .p-Widget.jp-RenderedJavaScript.jp-mod-trusted.jp-OutputArea-output:empty {\n",
       "              padding: 0;\n",
       "              border: 0;\n",
       "            }\n",
       "            </style>\n",
       "            <link rel=\"stylesheet\" href=\"https://cdn.datatables.net/1.10.20/css/jquery.dataTables.min.css\">\n",
       "            <table id=\"table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\" class=\"display\" style=\"display:block\"></table>\n",
       "            <script>\n",
       "              \n",
       "        if (typeof window.interactive_beam_jquery == 'undefined') {\n",
       "          var jqueryScript = document.createElement('script');\n",
       "          jqueryScript.src = 'https://code.jquery.com/jquery-3.4.1.slim.min.js';\n",
       "          jqueryScript.type = 'text/javascript';\n",
       "          jqueryScript.onload = function() {\n",
       "            var datatableScript = document.createElement('script');\n",
       "            datatableScript.src = 'https://cdn.datatables.net/1.10.20/js/jquery.dataTables.min.js';\n",
       "            datatableScript.type = 'text/javascript';\n",
       "            datatableScript.onload = function() {\n",
       "              window.interactive_beam_jquery = jQuery.noConflict(true);\n",
       "              window.interactive_beam_jquery(document).ready(function($){\n",
       "                \n",
       "            var dt;\n",
       "            if ($.fn.dataTable.isDataTable(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\")) {\n",
       "              dt = $(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\").dataTable();\n",
       "            } else if ($(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a_wrapper\").length == 0) {\n",
       "              dt = $(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\").dataTable({\n",
       "                \n",
       "            bAutoWidth: false,\n",
       "            columns: [{'title': ''}, {'title': 'window_start'}, {'title': 'window_end'}, {'title': 'page_views'}],\n",
       "            destroy: true,\n",
       "            responsive: true,\n",
       "            columnDefs: [\n",
       "              {\n",
       "                targets: \"_all\",\n",
       "                className: \"dt-left\"\n",
       "              },\n",
       "              {\n",
       "                \"targets\": 0,\n",
       "                \"width\": \"10px\",\n",
       "                \"title\": \"\"\n",
       "              }\n",
       "            ]\n",
       "              });\n",
       "            } else {\n",
       "              return;\n",
       "            }\n",
       "            dt.api()\n",
       "              .clear()\n",
       "              .rows.add([{1: '2024-04-01T17:30:00', 2: '2024-04-01T17:31:00', 3: '1963', 0: 0}, {1: '2024-04-01T17:28:00', 2: '2024-04-01T17:29:00', 3: '2023', 0: 1}, {1: '2024-04-01T17:27:00', 2: '2024-04-01T17:28:00', 3: '1899', 0: 2}, {1: '2024-04-01T17:29:00', 2: '2024-04-01T17:30:00', 3: '2050', 0: 3}, {1: '2024-04-01T17:31:00', 2: '2024-04-01T17:32:00', 3: '1970', 0: 4}, {1: '2024-04-01T17:32:00', 2: '2024-04-01T17:33:00', 3: '95', 0: 5}])\n",
       "              .draw('full-hold');\n",
       "              });\n",
       "            }\n",
       "            document.head.appendChild(datatableScript);\n",
       "          };\n",
       "          document.head.appendChild(jqueryScript);\n",
       "        } else {\n",
       "          window.interactive_beam_jquery(document).ready(function($){\n",
       "            \n",
       "            var dt;\n",
       "            if ($.fn.dataTable.isDataTable(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\")) {\n",
       "              dt = $(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\").dataTable();\n",
       "            } else if ($(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a_wrapper\").length == 0) {\n",
       "              dt = $(\"#table_df_7d02b74af8a1bf6bd7394e7ac9ec837a\").dataTable({\n",
       "                \n",
       "            bAutoWidth: false,\n",
       "            columns: [{'title': ''}, {'title': 'window_start'}, {'title': 'window_end'}, {'title': 'page_views'}],\n",
       "            destroy: true,\n",
       "            responsive: true,\n",
       "            columnDefs: [\n",
       "              {\n",
       "                targets: \"_all\",\n",
       "                className: \"dt-left\"\n",
       "              },\n",
       "              {\n",
       "                \"targets\": 0,\n",
       "                \"width\": \"10px\",\n",
       "                \"title\": \"\"\n",
       "              }\n",
       "            ]\n",
       "              });\n",
       "            } else {\n",
       "              return;\n",
       "            }\n",
       "            dt.api()\n",
       "              .clear()\n",
       "              .rows.add([{1: '2024-04-01T17:30:00', 2: '2024-04-01T17:31:00', 3: '1963', 0: 0}, {1: '2024-04-01T17:28:00', 2: '2024-04-01T17:29:00', 3: '2023', 0: 1}, {1: '2024-04-01T17:27:00', 2: '2024-04-01T17:28:00', 3: '1899', 0: 2}, {1: '2024-04-01T17:29:00', 2: '2024-04-01T17:30:00', 3: '2050', 0: 3}, {1: '2024-04-01T17:31:00', 2: '2024-04-01T17:32:00', 3: '1970', 0: 4}, {1: '2024-04-01T17:32:00', 2: '2024-04-01T17:33:00', 3: '95', 0: 5}])\n",
       "              .draw('full-hold');\n",
       "          });\n",
       "        }\n",
       "            </script>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/javascript": [
       "\n",
       "        if (typeof window.interactive_beam_jquery == 'undefined') {\n",
       "          var jqueryScript = document.createElement('script');\n",
       "          jqueryScript.src = 'https://code.jquery.com/jquery-3.4.1.slim.min.js';\n",
       "          jqueryScript.type = 'text/javascript';\n",
       "          jqueryScript.onload = function() {\n",
       "            var datatableScript = document.createElement('script');\n",
       "            datatableScript.src = 'https://cdn.datatables.net/1.10.20/js/jquery.dataTables.min.js';\n",
       "            datatableScript.type = 'text/javascript';\n",
       "            datatableScript.onload = function() {\n",
       "              window.interactive_beam_jquery = jQuery.noConflict(true);\n",
       "              window.interactive_beam_jquery(document).ready(function($){\n",
       "                \n",
       "            $(\"#progress_indicator_edfb8bc471267ecdc4353329135e82e5\").remove();\n",
       "              });\n",
       "            }\n",
       "            document.head.appendChild(datatableScript);\n",
       "          };\n",
       "          document.head.appendChild(jqueryScript);\n",
       "        } else {\n",
       "          window.interactive_beam_jquery(document).ready(function($){\n",
       "            \n",
       "            $(\"#progress_indicator_edfb8bc471267ecdc4353329135e82e5\").remove();\n",
       "          });\n",
       "        }"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "p = beam.Pipeline(interactive_runner.InteractiveRunner())\n",
    "\n",
    "events = (\n",
    "    p\n",
    "    | \"Read from files\"\n",
    "    >> beam.io.ReadFromText(\n",
    "        file_pattern=os.path.join(os.path.join(PARENT_DIR, \"inputs\", \"*.out\"))\n",
    "    )\n",
    "    | \"Parse elements\" >> beam.Map(parse_json).with_output_types(EventLog)\n",
    "    | \"Add event timestamp\" >> beam.Map(add_timestamp)\n",
    "    | \"Tumble window per minute\" >> beam.WindowInto(beam.window.FixedWindows(60))\n",
    "    | \"Count per minute\" >> beam.CombineGlobally(CountCombineFn()).without_defaults()\n",
    "    | \"Add window timestamp\" >> beam.ParDo(AddWindowTS())\n",
    ")\n",
    "\n",
    "ib.show(events)"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}