examples/07_end_to_end/wikimedia.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9eae9c30-99ce-4538-ab94-046515922a5f",
   "metadata": {},
   "source": [
    "# Adapters and Wikipedia data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "775cfbc0-099d-4328-9555-d351cbf3423b",
   "metadata": {},
   "source": [
    "In this example, we'll explore and visualize a set of data from Wikimedia displaying the most recent changes to any of the pages or files within Wikimedia domains using the CSP and Perspective libraries."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8c95ad84-b249-4810-97d8-b598228b3b7a",
   "metadata": {},
   "source": [
    "To do that, we need to install a few extra dependencies, listed below:"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e87d1d73-ba96-4cc0-96ad-7989461b568f",
   "metadata": {},
   "source": [
    "```\n",
    "pip install sseclient\n",
    "pip install perspective-python\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a617612f-ce10-4969-88cb-5f6fced13bfb",
   "metadata": {},
   "source": [
    "Next, we will need to write a new CSP [Adapter](https://github.com/Point72/csp/wiki/Adapters) to fetch the data and process it. In this example, we will write both a [Realtime Adapter](https://github.com/Point72/csp/wiki/Write-Realtime-Input-Adapters) and a [Historical Adapter](https://github.com/Point72/csp/wiki/Write-Historical-Input-Adapters)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08447511-b105-416f-a959-285330d417ba",
   "metadata": {},
   "source": [
    "**Note**: This tutorial has been tested for Python 3.11, using JupyterLab 4.2.0 and perspective-python 3.1.5."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1ecddaf1-4e1a-43c7-ab19-41eb5b670e49",
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip list | grep \"jupyterlab\" && pip list | grep \"perspective-python\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8329b504-5699-48e4-9b14-fe647fccf48a",
   "metadata": {},
   "source": [
    "## Writing adapters"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5cfef5ff-9ddc-47cb-be18-ddb2d14b7446",
   "metadata": {},
   "source": [
    "There are two main categories of writing input adapters, historical and realtime. When writing historical adapters you will need to implement a \"pull\" adapter, which pulls data from a historical data source in time order, one event at a time. When writing realtime adapters, you will need to implement a \"push\" adapter, which will get data from a separate thread that drives external events and \"pushes\" them into the engine as they occur.\n",
    "\n",
    "When writing input adapters it is also very important to denote the difference between \"graph building time\" and \"runtime\" versions of your adapter. Graph build time components solely describe the adapter. They are meant to do little else than keep track of the type of adapter and its parameters, which will then be used to construct the actual adapter implementation when the engine is constructed from the graph description. It is the runtime implementation that actual runs during the engine execution phase to process data."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7922e29d-ebc5-443c-a0f3-7665a267f2c0",
   "metadata": {},
   "source": [
    "## Using CSP to read realtime data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "580a80bf-b012-4c5f-b5d9-e12a1a2d3909",
   "metadata": {},
   "source": [
    "The data we are going to explore is the [MediaWiki Recent Changes feed](https://www.mediawiki.org/wiki/Manual:RCFeed) stream, which emits events related to recent changes across Wikimedia sites. The stream can be accessed through the [EventStreams web service](https://wikitech.wikimedia.org/wiki/Event_Platform/EventStreams_HTTP_Service)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c4633f0-e2de-4bb0-9828-c6089b832e21",
   "metadata": {},
   "source": [
    "Initially, we'll set the URL and define the data structure we expect to encounter."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c4008a7b-36d6-4d9a-accc-379fca61236d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime, timedelta\n",
    "\n",
    "URL = 'https://stream.wikimedia.org/v2/stream/recentchange'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fcdde9db-2273-4ac9-998c-d4f0d0b8c5a4",
   "metadata": {},
   "source": [
    "Next, we'll set up an initial (empty) visualization widget from the Perspective library so we can visualize the streaming data later. (Note that if you are running this example locally on JupyterLab, you may need to restart the Jupyter server after installing the Perspective library in order to visualize the widget.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ad33d769-c1a2-439b-be02-7abc7faccfdd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from perspective.widget import PerspectiveWidget\n",
    "\n",
    "# Data schema\n",
    "data = {\"servername\": \"string\", \"timestamp\": \"datetime\", \"event\": \"string\", \"servername_count\": \"integer\"}\n",
    "#widget = PerspectiveWidget(data, plugin=\"X Bar\", group_by=[\"servername\"], columns=[\"servername_count\"], aggregates={\"servername_count\": \"last\"}, theme='Pro Light', binding_mode=\"client-server\")\n",
    "widget = PerspectiveWidget(data,  plugin=\"X Bar\", group_by=[\"servername\"], columns=[\"servername_count\"], aggregates={\"servername_count\": \"last\"}, theme='Pro Light', binding_mode=\"client-server\")\n",
    "widget"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57481e87-c37b-4c89-a05e-7fc93abf8a56",
   "metadata": {},
   "source": [
    "Now, we will use CSP to build our EventSource adapter and update the Perspective widget we defined above with real-time events from the stream.\n",
    "\n",
    "First, we'll define a `csp.Struct` to hold our data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "967d0173-7688-46f0-b90c-0c4f91ee4dd1",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import re\n",
    "import json\n",
    "import threading\n",
    "\n",
    "from sseclient import SSEClient as EventSource\n",
    "\n",
    "import csp\n",
    "\n",
    "class WikiData(csp.Struct):\n",
    "    servername: str\n",
    "    servername_count: int\n",
    "    timestamp: str\n",
    "    event: str"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18dc5217-4732-4159-9a4c-5fd385184c74",
   "metadata": {},
   "source": [
    "Next, we will use a [PushInputAdapter](https://github.com/Point72/csp/wiki/Write-Realtime-Input-Adapters#pushinputadapter---python). This `PushInputAdapter` that you define will be used as the _runtime implementation_, and you also need to define a _graph- time_ representation of the time series edge by using `py_push_adapter_def`. \n",
    "\n",
    "When running the cell below, you should see the widget updated in real time with each new event, indexed by the server where the update happened."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f983793a-7785-4759-ad8a-80b9f0bd8331",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from csp.impl.pushadapter import PushInputAdapter\n",
    "from csp.impl.wiring import py_push_adapter_def\n",
    "\n",
    "# Define the runtime implementation of our adapter\n",
    "class FetchWikiDataAdapter(PushInputAdapter):\n",
    "    def __init__(self, url: str):\n",
    "        self._thread = None\n",
    "        self._running = False\n",
    "        self._url = url\n",
    "\n",
    "    def start(self, starttime, endtime):\n",
    "        print(\"FetchWikiDataAdapter::start\")\n",
    "        self._source = EventSource(self._url)\n",
    "        self._running = True\n",
    "        self._thread = threading.Thread(target=self._run)\n",
    "        self._thread.start()\n",
    "\n",
    "    def stop(self):\n",
    "        print(\"FetchWikiDataAdapter::stop\")\n",
    "        if self._running:\n",
    "            self._running = False\n",
    "            self._thread.join()\n",
    "            self._source.resp.close()\n",
    "\n",
    "    def _run(self):\n",
    "        servernames = dict([])\n",
    "        for item in self._source:\n",
    "            if not self._running:\n",
    "                break\n",
    "            if item.event == 'message':\n",
    "                try:\n",
    "                    change = json.loads(item.data)\n",
    "                except ValueError:\n",
    "                    pass\n",
    "                else:\n",
    "                    # discard canary events\n",
    "                    # WMF Data Engineering team produces artificial 'canary' events into \n",
    "                    # each stream multiple times an hour. The presence of these canary\n",
    "                    # events in a stream allow us to differentiate between a broken event\n",
    "                    # stream, and an empty one.\n",
    "                    # We will also filter bot-generated events\n",
    "                    if change['meta']['domain'] == 'canary' or re.search('bot', change['user'], re.IGNORECASE):\n",
    "                        continue\n",
    "                    timestamp = change['meta']['dt']\n",
    "                    event = f\"{timestamp}:: {change['user']} edited {change['title']}\"\n",
    "                    servername = change['server_name']\n",
    "                    # Manually \"tick\" this edge\n",
    "                    self.push_tick(\n",
    "                        WikiData(\n",
    "                            servername=servername,\n",
    "                            servername_count=0,\n",
    "                            timestamp=timestamp,\n",
    "                            event=event\n",
    "                        )\n",
    "                    )\n",
    "\n",
    "# Create the graph-time representation of our adapter\n",
    "FetchWikiData = py_push_adapter_def(\"FetchWikiData\", FetchWikiDataAdapter, csp.ts[WikiData], url=str)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af924b1b-3734-42c8-8978-79f5c88ffb54",
   "metadata": {},
   "source": [
    "Now, we define the graph and run the engine."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9f105395-b9d5-4fa4-b752-48e6d2618fae",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "@csp.node\n",
    "def update_widget(wiki_event: csp.ts[WikiData], widget: PerspectiveWidget, throttle: timedelta = timedelta(seconds=0.5)):\n",
    "    # Updates the perspective widget with batched updates for scalability\n",
    "    with csp.alarms():\n",
    "        alarm = csp.alarm(bool)\n",
    "\n",
    "    with csp.state():\n",
    "        s_buffer = []\n",
    "\n",
    "    with csp.start():\n",
    "        csp.schedule_alarm(alarm, throttle, True)\n",
    "        \n",
    "    if csp.ticked(wiki_event):\n",
    "        s_buffer.append({\n",
    "            \"servername\": wiki_event.servername,\n",
    "            \"servername_count\": wiki_event.servername_count,\n",
    "            \"timestamp\": wiki_event.timestamp,\n",
    "            \"event\": wiki_event.event,\n",
    "        })\n",
    "\n",
    "    if csp.ticked(alarm):\n",
    "        if len(s_buffer) > 0:\n",
    "            widget.update(s_buffer)\n",
    "            s_buffer = []\n",
    "\n",
    "        csp.schedule_alarm(alarm, throttle, True)\n",
    "\n",
    "@csp.node\n",
    "def compute_server_count(wiki_event: csp.ts[WikiData]) -> csp.ts[WikiData]:\n",
    "    # takes the raw struct in, creates a copy with the count set and ticks it out\n",
    "    with csp.state():\n",
    "        s_servernames = dict([])\n",
    "    if csp.ticked(wiki_event):\n",
    "        if wiki_event.servername in s_servernames:\n",
    "            s_servernames[wiki_event.servername] += 1\n",
    "        else:\n",
    "            s_servernames[wiki_event.servername] = 1\n",
    "    wiki_event.servername_count = s_servernames[wiki_event.servername]\n",
    "    return wiki_event\n",
    "\n",
    "@csp.graph\n",
    "def wiki_graph():\n",
    "    print(\"Start of graph building\")\n",
    "    URL = 'https://stream.wikimedia.org/v2/stream/recentchange'\n",
    "    # Create an edge in the graph with the events fetched by the adapter\n",
    "    result = FetchWikiData(url=URL)\n",
    "    result = compute_server_count(result)\n",
    "    # Update Perspective widget with each event/tick\n",
    "    update_widget(result, widget=widget)\n",
    "    # We can also print the event object\n",
    "    csp.print(\"Event\", result.event)\n",
    "    # Add this edge as a graph output\n",
    "    csp.add_graph_output(\"Wiki events\", result.event)\n",
    "    print(\"End of graph building\")\n",
    "\n",
    "start = datetime.utcnow()\n",
    "csp.run(wiki_graph, starttime=start, endtime=start+timedelta(seconds=30), realtime=True)\n",
    "print(\"Done.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f707e182-df6e-4802-af78-3f691f0fdf42",
   "metadata": {},
   "source": [
    "## Using CSP to read historical data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4104ae2f-defd-41a3-85ae-30f2c2271b3e",
   "metadata": {},
   "source": [
    "Now, let's explore writing a _historical_ adapter to read the same data stream. The Wikimedia API offers the following query for historical data:\n",
    "\n",
    "```\n",
    "URL = f'https://stream.wikimedia.org/v2/stream/recentchange?since={start_date}'\n",
    "```\n",
    "\n",
    "where `{start_date}` is a string representing the start of the period we want to read from. \n",
    "\n",
    "The stream history for Wikimedia data is not kept indefinitely. Depending on the particular stream configuration, there will likely be between 7 and 31 days of history available. We will choose a past date to start reading data, and stop after 23h of events."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7bc632d9-b47c-4f3d-87fc-3364f90dab4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import timezone\n",
    "\n",
    "start_time = datetime.now() - timedelta(days=2)\n",
    "end_time = start_time.replace(tzinfo=timezone.utc) + timedelta(hours=23)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1ee1bf33-50f8-4256-a003-f38077f888cc",
   "metadata": {},
   "source": [
    "To read this data, we will use a [Pull input adapter](https://github.com/Point72/csp/wiki/Write-Historical-Input-Adapters#pullinputadapter)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9aee0073-6595-47b4-8144-45b6c417af65",
   "metadata": {},
   "outputs": [],
   "source": [
    "from csp.impl.pulladapter import PullInputAdapter\n",
    "from csp.impl.wiring import py_pull_adapter_def\n",
    "\n",
    "# Define the runtime implementation of our adapter\n",
    "class HistoricalWikiDataAdapter(PullInputAdapter):\n",
    "    def __init__(self, url: str):\n",
    "        self._url = url\n",
    "        self._events = []\n",
    "        super().__init__()\n",
    "\n",
    "    def start(self, starttime, endtime):\n",
    "        print(\"HistoricalWikiDataAdapter::start\")\n",
    "        self._events = EventSource(self._url)\n",
    "\n",
    "    def stop(self):\n",
    "        print(\"HistoricalWikiDataAdapter::stop\")\n",
    "    \n",
    "    def next(self):\n",
    "        \"\"\"Return tuple of datetime, value of next tick, or None if no more data is available\"\"\"\n",
    "        self._next_event = next(self._events)\n",
    "        time = self._start_time\n",
    "        if self._next_event and time < self._end_time:\n",
    "             while True:\n",
    "                item = self._next_event\n",
    "                if item.event == 'message':\n",
    "                    try:\n",
    "                        change = json.loads(item.data)            \n",
    "                    except ValueError:\n",
    "                        self._next_event = next(self._events)\n",
    "                        continue\n",
    "                    else:\n",
    "                        # discard canary events\n",
    "                        # WMF Data Engineering team produces artificial 'canary' events into \n",
    "                        # each stream multiple times an hour. The presence of these canary\n",
    "                        # events in a stream allow us to differentiate between a broken event\n",
    "                        # stream, and an empty one.\n",
    "                        # We will also filter bot-generated events\n",
    "                        if change['meta']['domain'] == 'canary' or re.search('bot', change['user'], re.IGNORECASE):\n",
    "                            self._next_event = next(self._events)\n",
    "                            continue\n",
    "                        timestamp = change['meta']['dt']\n",
    "                        time = datetime.fromisoformat(timestamp.rstrip('Z'))\n",
    "                        if time <= self._end_time:\n",
    "                            event = f\"{timestamp}:: {change['user']} edited {change['title']}\"\n",
    "                            servername = change['server_name']\n",
    "                            # Manually \"tick\" this edge\n",
    "                            return (\n",
    "                                time,\n",
    "                                WikiData(\n",
    "                                    servername=servername,\n",
    "                                    timestamp=timestamp,\n",
    "                                    event=event,\n",
    "                                    servername_count=0,\n",
    "                                )\n",
    "                            )\n",
    "        return None\n",
    "\n",
    "# Create the graph-time representation of our adapter\n",
    "HistoricalWikiData = py_pull_adapter_def(\n",
    "    \"HistoricalWikiData\",\n",
    "    HistoricalWikiDataAdapter,\n",
    "    csp.ts[WikiData],\n",
    "    url=str\n",
    ")\n",
    "\n",
    "@csp.node\n",
    "def pretty_print(event: csp.ts[WikiData]) -> csp.ts[str]:\n",
    "    if csp.ticked(event):\n",
    "        return f\"{event.event}\"\n",
    "\n",
    "@csp.graph\n",
    "def wiki_graph():\n",
    "    URL = f\"https://stream.wikimedia.org/v2/stream/recentchange?since={start_time}\"\n",
    "    print(f\"Start of graph building from {start_time} until {end_time}\")\n",
    "    # Create an edge in the graph with the events fetched by the adapter\n",
    "    events = HistoricalWikiData(url=URL)\n",
    "    # We will process the events data through a node, and only print events\n",
    "    # happening at en.wikipedia.org\n",
    "    en_wiki = csp.filter(events.servername == \"en.wikipedia.org\", events)\n",
    "    # Let's pretty-print the results\n",
    "    en_events = pretty_print(en_wiki)\n",
    "    csp.print(\"Wiki event:\", en_events)\n",
    "    # Add this edge as a graph output\n",
    "    # This allows you to connect an edge as a \"graph output\", returned to the\n",
    "    # caller from csp.run as a dictionary of key: [(datetime, value)]\n",
    "    csp.add_graph_output(\"en_wiki\", en_wiki)\n",
    "    print(\"End of graph building\")\n",
    "\n",
    "csp.run(wiki_graph, starttime=start_time, endtime=end_time)\n",
    "print(\"Done.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9644014-21b9-46d3-af94-8f2e5615f04a",
   "metadata": {},
   "source": [
    "The strongest point of CSP is that we can use the historical edge or the realtime edge interchangeably for the same graph, as shown below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "06f153b7-89fb-4d7b-b562-9cefc37024d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "@csp.node\n",
    "def pretty_print(event: csp.ts[WikiData]) -> csp.ts[str]:\n",
    "    if csp.ticked(event):\n",
    "        return f\"{event.event}\"\n",
    "\n",
    "@csp.graph\n",
    "def wiki_graph():\n",
    "    print(\"Start of graph building\")\n",
    "    if csp.is_configured_realtime():\n",
    "        URL = \"https://stream.wikimedia.org/v2/stream/recentchange\"\n",
    "        events = FetchWikiData(url=URL)\n",
    "    else:\n",
    "        URL = f\"https://stream.wikimedia.org/v2/stream/recentchange?since={start_time}\"\n",
    "        events = HistoricalWikiData(url=URL)\n",
    "\n",
    "    # We will process the events data through a node, and only print events\n",
    "    # happening at en.wikipedia.org\n",
    "    en_wiki = csp.filter(events.servername == \"en.wikipedia.org\", events)\n",
    "    # Let's pretty-print the results\n",
    "    en_events = pretty_print(en_wiki)\n",
    "    csp.print(\"Wiki event:\", en_events)\n",
    "    # Add this edge as a graph output\n",
    "    # This allows you to connect an edge as a \"graph output\", returned to the\n",
    "    # caller from csp.run as a dictionary of key: [(datetime, value)]\n",
    "    csp.add_graph_output(\"en_wiki\", en_wiki)\n",
    "    print(\"End of graph building\")\n",
    "\n",
    "start_time = datetime.utcnow()\n",
    "end_time = start_time+timedelta(seconds=30)\n",
    "csp.run(wiki_graph, starttime=start_time, endtime=end_time, realtime=True)\n",
    "print(\"Done.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c6228a65-7cd4-455c-98ce-4e3daa303e99",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34424d83-a2e4-4e39-b333-cce2c333f1d0",
   "metadata": {},
   "source": [
    "### References\n",
    "\n",
    "1. https://github.com/Point72/csp/wiki/5.-Adapters#writing-input-and-output-adapters\n",
    "2. https://wikitech.wikimedia.org/wiki/Event_Platform/EventStreams\n",
    "3. https://schema.wikimedia.org/repositories/primary/jsonschema/mediawiki/recentchange/latest.yaml\n",
    "4. https://github.com/finos/perspective"
   ]
  }
 ],
 "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.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}