T301902-get-events.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e5cc00d3-147a-45b5-98e2-cc487f5b402d",
   "metadata": {},
   "source": [
    "# Search Events Data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fb5406d5-5034-4791-9476-1127b81b634a",
   "metadata": {},
   "source": [
    "[T301902](https://phabricator.wikimedia.org/T301902)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c8b96154-cf72-4b74-8ae7-8475e342baca",
   "metadata": {},
   "source": [
    "This notebook is for pulling reduced search event data from `searchsatisfaction` table.\n",
    "\n",
    "We are interested in the following emerging languages for the search experimentations:\n",
    "\n",
    "Priority 1:\n",
    "Arabic, Bengali*, Spanish, Portuguese*, Russian\n",
    "\n",
    "Priority 2: French*, Korean*, Indonesian, Ukrainian, Thai* ,Malaysian (?), Hindi, Tagalog, Afrikaans, Cantonese, Malayalam, Telugu"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "30873f08-e9d7-4a74-9c81-34979d4322c1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "You are using wmfdata v1.3.1, but v1.3.3 is available.\n",
      "\n",
      "To update, run `pip install --upgrade git+https://github.com/wikimedia/wmfdata-python.git@release --ignore-installed`.\n",
      "\n",
      "To see the changes, refer to https://github.com/wikimedia/wmfdata-python/blob/release/CHANGELOG.md\n"
     ]
    }
   ],
   "source": [
    "import datetime as dt\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "from wmfdata import hive, spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e04a419b-df8c-460c-b32a-3c81031e9e0a",
   "metadata": {},
   "source": [
    "## Notes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7bc80fc9-9b0a-4682-8c1e-aa543946d97b",
   "metadata": {},
   "source": [
    "Timestamps: we'll coalesce `dt` and `meta.dt`, and trust whatever comes out of it. When I investigated timestamps for searches on Commons I found that there are peaks around the various hour intervals, but they're incredibly small compared to the correct timestamp. To begin with, it's easier to trust these timestamps than develop heuristics to change them.\n",
    "\n",
    "Event logging in SearchSatisfaction is only done on the desktop platform. \n",
    "TO DO: We will work with web team to rebuild search database for mobile web searches. \n",
    "\n",
    "Users who have Do Not Track enabled are not part of the dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "f99aa598-173e-4a17-a48b-975b4b64eb55",
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki = \"'ruwiki', 'arwiki', 'bnwiki', 'eswiki', 'ptwiki'\"\n",
    "wiki2 = \"'frwiki', 'kowiki', 'idwiki', 'ukwiki', 'thwiki', 'mswiki', 'hiwiki', 'tlwiki', 'afwiki', 'zh-yuewiki', 'mlwiki', 'tewiki'\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8a397fda-c2f8-490b-b00d-f9538f499ce5",
   "metadata": {},
   "source": [
    "## Configuring Timestamps"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c70439b5-e167-47d7-bb88-70a4b2dd3e14",
   "metadata": {},
   "source": [
    "Configuring Timestamps\n",
    "We'll call the day we're gathering data for `data_day`. We're also expecting this notebook to be run the day after, which we'll call `next_day`. In order to ignore search sessions that started on the previous day, we also define that day. Lastly, we set a limit of one hour after midnight UTC as the cutoff for data. In other words, we expect search sessions to be completed within one hour."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "8c88e36c-0027-48c4-9f26-621f64564749",
   "metadata": {},
   "outputs": [],
   "source": [
    "next_day = dt.datetime.now(dt.timezone.utc).date()\n",
    "\n",
    "data_day = next_day - dt.timedelta(days = 1)\n",
    "previous_day = data_day - dt.timedelta(days = 1)\n",
    "\n",
    "limit_timestamp = dt.datetime.combine(next_day, dt.time(hour = 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7d8a8a2a-69fc-4d00-980b-b806e902e6cd",
   "metadata": {},
   "source": [
    "## Create Aggregation Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "bc0d930e-8e41-4469-bbcc-4ce49483f90c",
   "metadata": {},
   "outputs": [],
   "source": [
    "event_table = 'cchen_search.search_events'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "c1e63642-f8c9-44ae-ba3b-b48a3abd3d5e",
   "metadata": {},
   "outputs": [],
   "source": [
    "create_table_query = '''\n",
    "CREATE TABLE IF NOT EXISTS {table_name} (\n",
    "    dt STRING,\n",
    "    wiki STRING,\n",
    "    session_id STRING,\n",
    "    unique_id STRING,\n",
    "    action STRING,\n",
    "    source STRING,\n",
    "    input_location STRING,\n",
    "    query STRING,\n",
    "    results_returned BIGINT,\n",
    "    click_position BIGINT,\n",
    "    pageview_id STRING,\n",
    "    user_is_bot BOOLEAN\n",
    ")\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5aae1e39-6164-4698-a03d-a9d9d78c5b58",
   "metadata": {},
   "outputs": [],
   "source": [
    "hive.run(create_table_query.format(\n",
    "            table_name = event_table\n",
    "))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88beed0e-3b7a-47e5-b7b8-ba2a5eec61a3",
   "metadata": {},
   "source": [
    "## Timestamp Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "0e8ead83-9faf-4a46-86e5-63362268726c",
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_partition_statement(start_ts, end_ts, prefix = ''):\n",
    "    '''\n",
    "    This takes the two timestamps and creates a statement that selects\n",
    "    partitions based on `year`, `month`, and `day` in order to make our\n",
    "    data gathering not use excessive amounts of data. It assumes that\n",
    "    `start_ts` and `end_ts` are not more than a month apart, which should\n",
    "    be a reasonable expectation for this notebook.\n",
    "    \n",
    "    An optional prefix can be set to enable selecting partitions for\n",
    "    multiple tables with different aliases.\n",
    "    \n",
    "    :param start_ts: start timestamp\n",
    "    :type start_ts: datetime.datetime\n",
    "    \n",
    "    :param end_ts: end timestamp\n",
    "    :type end_ts: datetime.datetime\n",
    "    \n",
    "    :param prefix: prefix to use in front of partition clauses, \".\" is added automatically\n",
    "    :type prefix: str\n",
    "    '''\n",
    "    \n",
    "    if prefix:\n",
    "        prefix = f'{prefix}.' # adds \".\" after the prefix\n",
    "    \n",
    "    # there are three cases:\n",
    "    # 1: month and year are the same, output a \"BETWEEN\" statement with the days\n",
    "    # 2: months differ, but the years are the same.\n",
    "    # 3: years differ too.\n",
    "    # Case #2 and #3 can be combined, because it doesn't really matter\n",
    "    # if the years are the same in the month-selection or not.\n",
    "    \n",
    "    if start_ts.year == end_ts.year and start_ts.month == end_ts.month:\n",
    "        return(f'''{prefix}year = {start_ts.year}\n",
    "AND {prefix}month = {start_ts.month}\n",
    "AND {prefix}day BETWEEN {start_ts.day} AND {end_ts.day}''')\n",
    "    else:\n",
    "        return(f'''\n",
    "(\n",
    "    ({prefix}year = {start_ts.year}\n",
    "     AND {prefix}month = {start_ts.month}\n",
    "     AND {prefix}day >= {start_ts.day})\n",
    " OR ({prefix}year = {end_ts.year}\n",
    "     AND {prefix}month = {end_ts.month}\n",
    "     AND {prefix}day <= {end_ts.day})\n",
    ")''')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "75e0950c-d3d9-4330-ae28-0459483c8d4e",
   "metadata": {},
   "source": [
    "## Get Event Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "8bc7a9e2-89d8-4953-87b6-96233e529c01",
   "metadata": {},
   "outputs": [],
   "source": [
    "event_query = ''' \n",
    "    \n",
    "    INSERT INTO cchen_search.search_events\n",
    "    \n",
    "    SELECT \n",
    "        MIN(coalesce(client_dt, meta.dt)) AS dt, \n",
    "        wiki AS wiki_db,\n",
    "        event.searchsessionid AS session_id,\n",
    "        event.uniqueid AS unique_id,\n",
    "        event.action AS action,\n",
    "        event.source AS source, \n",
    "        event.inputlocation AS input_location,\n",
    "        event.query AS query,\n",
    "        event.hitsReturned AS results_returned,\n",
    "        event.position AS click_position,\n",
    "        event.pageviewid AS pageview_id,\n",
    "        useragent.is_bot AS user_is_bot\n",
    "    FROM event.searchsatisfaction ess\n",
    "    WHERE\n",
    "        {ess_partition_statement}\n",
    "        AND wiki  in ({wiki_db})\n",
    "        AND event.subTest IS NULL\n",
    "        AND event.isforced IS NULL -- only include non-test users\n",
    "    GROUP BY \n",
    "        wiki,\n",
    "        event.searchsessionid ,\n",
    "        event.uniqueid ,\n",
    "        event.action ,\n",
    "        event.source , \n",
    "        event.inputlocation ,\n",
    "        event.query ,\n",
    "        event.hitsReturned , \n",
    "        event.position,\n",
    "        event.pageviewid,\n",
    "        useragent.is_bot\n",
    "    HAVING\n",
    "        TO_DATE(dt) = '{today}'\n",
    "        \n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "38bbf774-6c76-4cd0-b90e-53eb1360aba0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# set up days\n",
    "first_day = dt.date(2022, 4, 1) \n",
    "last_day = dt.date(2022, 4, 11)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "243ea40b-2bbd-40da-9a78-a2a3d6a69616",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-03-31 (simulating cron job on 2022-04-01)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-01 (simulating cron job on 2022-04-02)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-02 (simulating cron job on 2022-04-03)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-03 (simulating cron job on 2022-04-04)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-04 (simulating cron job on 2022-04-05)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-05 (simulating cron job on 2022-04-06)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-06 (simulating cron job on 2022-04-07)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-07 (simulating cron job on 2022-04-08)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-08 (simulating cron job on 2022-04-09)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-09 (simulating cron job on 2022-04-10)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "running data gathering for 2022-04-10 (simulating cron job on 2022-04-11)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "PySpark executors will use /usr/lib/anaconda-wmf/bin/python3.\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "current_day = first_day\n",
    "\n",
    "while current_day <= last_day:\n",
    "    # calculate days\n",
    "    next_day = current_day\n",
    "    data_day = next_day - dt.timedelta(days = 1)\n",
    "    previous_day = data_day - dt.timedelta(days = 1)\n",
    "\n",
    "    limit_timestamp = dt.datetime.combine(next_day, dt.time(hour = 1))\n",
    "    \n",
    "    # print some helpful stuff\n",
    "    print(f'running data gathering for {data_day} (simulating cron job on {current_day})')\n",
    "    \n",
    "    try:\n",
    "        spark.run(event_query.format(\n",
    "            today = data_day,\n",
    "            limit_timestamp = limit_timestamp.isoformat(),\n",
    "            ess_partition_statement = make_partition_statement(previous_day, next_day, prefix = 'ess'),\n",
    "            wiki_db = wiki\n",
    "        ))\n",
    "    except UnboundLocalError:\n",
    "        pass\n",
    "    \n",
    "    current_day += dt.timedelta(days = 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9f3e4e7e-9fa0-433f-8526-5bc538c9c8bd",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}