Threshhold_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 247,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Window\n",
    "from functools import reduce\n",
    "from pyspark.sql.functions import pandas_udf, PandasUDFType\n",
    "from pyspark.sql.functions import udf\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 248,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 249,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/mwcomments-0.3.3-py3.5.egg\")\n",
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/sortedcontainers-2.1.0-py3.5.egg\")\n",
    "spark.sparkContext.addPyFile(\"../.local/lib/python3.5/site-packages/python_dateutil-2.8.0-py3.5.egg\")\n",
    "spark.sparkContext.addPyFile(\"./spark_functions.py\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 250,
   "metadata": {},
   "outputs": [],
   "source": [
    "import mwcomments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 251,
   "metadata": {},
   "outputs": [],
   "source": [
    "from project_settings import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 252,
   "metadata": {},
   "outputs": [],
   "source": [
    "remeber_dict = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 253,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs = pd.read_csv(os.path.join(data_dir,\"ores_rcfilters_cutoffs.csv\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 254,
   "metadata": {},
   "outputs": [],
   "source": [
    "# compare 14 days before and after the cutoff\n",
    "# unless there's another cutoff less than 28 days away, in which case split the difference\n",
    "by_wiki = cutoffs.groupby('wiki_db')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 255,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs['date'] = pd.to_datetime(cutoffs.deploy_dt)\n",
    "cutoffs = cutoffs.drop(\"deploy_gap\",1)\n",
    "cutoffs = cutoffs.drop(\"deploy_dt\",1)\n",
    "cutoffs = cutoffs.drop(\"commit_dt\",1)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 256,
   "metadata": {},
   "outputs": [],
   "source": [
    "def set_cutoff_period(df):\n",
    "    df = df.sort_values(by=['date'])\n",
    "    next_cutoff = df.shift(1)\n",
    "    df['time_since_last_cutoff']  = df.date - df.shift(1).date\n",
    "    df['time_till_next_cutoff']  = df.shift(-1).date - df.date\n",
    "    df = df.reset_index()\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 257,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs = cutoffs.groupby(\"wiki_db\").apply(set_cutoff_period)\n",
    "\n",
    "cutoffs = cutoffs.drop('wiki_db',1).reset_index()\n",
    "cutoffs = cutoffs.drop(\"level_1\",1)\n",
    "\n",
    "select =[ 'wiki_db','has_ores','has_rcfilters','has_rcfilters_watchlist','time_since_last_cutoff','time_till_next_cutoff','date']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 258,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We special case wikis where some issues lead to changes and deployments that we don't want to analyze. \n",
    "# fawiki: bug leads to cutoff disabling ores for 2 days. These won't show up in any other interval, so ignore them. \n",
    "cutoffs = cutoffs.loc[~((cutoffs.wiki_db == 'fawiki') & ( (cutoffs.date == pd.to_datetime(\"2017-12-09 11:19:00\")) | (cutoffs.date == pd.to_datetime(\"2017-12-11 18:56:00\"))))]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 259,
   "metadata": {},
   "outputs": [],
   "source": [
    "# etwiki, frwiki, and hewiki apparently turned on rcfilters 50 days after enabling ORES. This is OK. The periods overlap.\n",
    "\n",
    "#frwiki and ruwiki experienced a bug on the deployment of rcfilters to watchlist. So let's ignore them for those messages.   \n",
    "\n",
    "cutoffs = cutoffs.loc[~((cutoffs.wiki_db == 'frwiki') & (cutoffs.date >= pd.to_datetime(\"2017-11-09 14:35:00\")))]\n",
    "\n",
    "cutoffs = cutoffs.loc[~((cutoffs.wiki_db == 'ruwiki') & (cutoffs.date >= pd.to_datetime(\"2017-11-20 19:22:00\") ))]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 260,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>wiki_db</th>\n",
       "      <th>has_ores</th>\n",
       "      <th>has_rcfilters</th>\n",
       "      <th>has_rcfilters_watchlist</th>\n",
       "      <th>time_since_last_cutoff</th>\n",
       "      <th>time_till_next_cutoff</th>\n",
       "      <th>date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>cswiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>76 days 23:21:00</td>\n",
       "      <td>2017-02-22 00:33:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>cswiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>76 days 23:21:00</td>\n",
       "      <td>308 days 17:24:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>etwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>50 days 05:26:00</td>\n",
       "      <td>2017-03-20 18:28:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>etwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>50 days 05:26:00</td>\n",
       "      <td>308 days 17:24:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>22</th>\n",
       "      <td>frwiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>55 days 07:02:00</td>\n",
       "      <td>2017-04-11 11:09:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>23</th>\n",
       "      <td>frwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>55 days 07:02:00</td>\n",
       "      <td>156 days 20:24:00</td>\n",
       "      <td>2017-06-05 18:11:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>27</th>\n",
       "      <td>hewiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>29 days 10:38:00</td>\n",
       "      <td>2017-04-10 13:16:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>28</th>\n",
       "      <td>hewiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>29 days 10:38:00</td>\n",
       "      <td>308 days 17:24:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>32</th>\n",
       "      <td>kowiki</td>\n",
       "      <td>False</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>2019-03-04 16:27:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>33</th>\n",
       "      <td>kowiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>NaT</td>\n",
       "      <td>2019-03-04 16:27:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>65</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>169 days 13:27:00</td>\n",
       "      <td>3 days 23:58:00</td>\n",
       "      <td>2017-10-26 13:21:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>66</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>3 days 23:58:00</td>\n",
       "      <td>28 days 05:52:00</td>\n",
       "      <td>2017-10-30 13:19:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>67</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>28 days 05:52:00</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>2017-11-27 19:11:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>68</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>NaT</td>\n",
       "      <td>2017-11-27 19:11:00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         wiki_db has_ores has_rcfilters has_rcfilters_watchlist  \\\n",
       "3         cswiki     True         False                   False   \n",
       "4         cswiki     True          True                   False   \n",
       "12        etwiki     True         False                   False   \n",
       "13        etwiki     True          True                   False   \n",
       "22        frwiki    False         False                   False   \n",
       "23        frwiki     True          True                   False   \n",
       "27        hewiki     True         False                   False   \n",
       "28        hewiki     True          True                   False   \n",
       "32        kowiki    False          True                   False   \n",
       "33        kowiki     True          True                    True   \n",
       "65  wikidatawiki    False         False                   False   \n",
       "66  wikidatawiki     True          True                    True   \n",
       "67  wikidatawiki    False         False                   False   \n",
       "68  wikidatawiki     True          True                    True   \n",
       "\n",
       "    time_since_last_cutoff  time_till_next_cutoff                date  \n",
       "3                      NaT       76 days 23:21:00 2017-02-22 00:33:00  \n",
       "4         76 days 23:21:00      308 days 17:24:00 2017-05-09 23:54:00  \n",
       "12                     NaT       50 days 05:26:00 2017-03-20 18:28:00  \n",
       "13        50 days 05:26:00      308 days 17:24:00 2017-05-09 23:54:00  \n",
       "22                     NaT       55 days 07:02:00 2017-04-11 11:09:00  \n",
       "23        55 days 07:02:00      156 days 20:24:00 2017-06-05 18:11:00  \n",
       "27                     NaT       29 days 10:38:00 2017-04-10 13:16:00  \n",
       "28        29 days 10:38:00      308 days 17:24:00 2017-05-09 23:54:00  \n",
       "32                     NaT        0 days 00:00:00 2019-03-04 16:27:00  \n",
       "33         0 days 00:00:00                    NaT 2019-03-04 16:27:00  \n",
       "65       169 days 13:27:00        3 days 23:58:00 2017-10-26 13:21:00  \n",
       "66         3 days 23:58:00       28 days 05:52:00 2017-10-30 13:19:00  \n",
       "67        28 days 05:52:00        0 days 00:00:00 2017-11-27 19:11:00  \n",
       "68         0 days 00:00:00                    NaT 2017-11-27 19:11:00  "
      ]
     },
     "execution_count": 260,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cutoffs.loc[(cutoffs.time_since_last_cutoff<=pd.Timedelta(120,'D')) | (cutoffs.time_till_next_cutoff<=pd.Timedelta(120,'D')), select]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 261,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>wiki_db</th>\n",
       "      <th>index</th>\n",
       "      <th>commitsha</th>\n",
       "      <th>damaging_hard</th>\n",
       "      <th>damaging_likelybad_max</th>\n",
       "      <th>damaging_likelybad_min</th>\n",
       "      <th>damaging_likelygood_max</th>\n",
       "      <th>damaging_likelygood_min</th>\n",
       "      <th>damaging_maybebad_max</th>\n",
       "      <th>damaging_maybebad_min</th>\n",
       "      <th>...</th>\n",
       "      <th>extension_status_default</th>\n",
       "      <th>rcfilters_watchlist_enabled_default</th>\n",
       "      <th>commitsha_default</th>\n",
       "      <th>has_extension</th>\n",
       "      <th>has_beta_extension</th>\n",
       "      <th>has_rcfilters</th>\n",
       "      <th>has_rcfilters_watchlist</th>\n",
       "      <th>date</th>\n",
       "      <th>time_since_last_cutoff</th>\n",
       "      <th>time_till_next_cutoff</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>arwiki</td>\n",
       "      <td>0</td>\n",
       "      <td>c19a7d1dd4f8b720ef99cd5a33a9853a0555fa1c</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>maximum recall @ precision &gt;= 0.45</td>\n",
       "      <td>maximum recall @ precision &gt;= 0.997</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>True</td>\n",
       "      <td>22f6ec8967a3b66019881f27bb37515eaae855bc</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>2018-05-09 18:04:00</td>\n",
       "      <td>NaT</td>\n",
       "      <td>NaT</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>1 rows × 43 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "  wiki_db  index                                 commitsha  damaging_hard  \\\n",
       "0  arwiki      0  c19a7d1dd4f8b720ef99cd5a33a9853a0555fa1c            NaN   \n",
       "\n",
       "   damaging_likelybad_max              damaging_likelybad_min  \\\n",
       "0                     1.0  maximum recall @ precision >= 0.45   \n",
       "\n",
       "               damaging_likelygood_max  damaging_likelygood_min  \\\n",
       "0  maximum recall @ precision >= 0.997                      0.0   \n",
       "\n",
       "   damaging_maybebad_max damaging_maybebad_min          ...           \\\n",
       "0                    NaN                   NaN          ...            \n",
       "\n",
       "   extension_status_default  rcfilters_watchlist_enabled_default  \\\n",
       "0                       NaN                                 True   \n",
       "\n",
       "                          commitsha_default has_extension has_beta_extension  \\\n",
       "0  22f6ec8967a3b66019881f27bb37515eaae855bc         False              False   \n",
       "\n",
       "   has_rcfilters  has_rcfilters_watchlist                date  \\\n",
       "0           True                     True 2018-05-09 18:04:00   \n",
       "\n",
       "   time_since_last_cutoff time_till_next_cutoff  \n",
       "0                     NaT                   NaT  \n",
       "\n",
       "[1 rows x 43 columns]"
      ]
     },
     "execution_count": 261,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cutoffs.loc[cutoffs.wiki_db == 'arwiki']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 262,
   "metadata": {},
   "outputs": [],
   "source": [
    "# kowiki was enabled over two commits with the same deploy time\n",
    "cutoffs = cutoffs.drop(32)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 263,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>wiki_db</th>\n",
       "      <th>has_ores</th>\n",
       "      <th>has_rcfilters</th>\n",
       "      <th>has_rcfilters_watchlist</th>\n",
       "      <th>time_since_last_cutoff</th>\n",
       "      <th>time_till_next_cutoff</th>\n",
       "      <th>date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>63</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>321 days 08:41:00</td>\n",
       "      <td>2016-06-22 15:13:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>64</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>321 days 08:41:00</td>\n",
       "      <td>169 days 13:27:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>65</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>169 days 13:27:00</td>\n",
       "      <td>3 days 23:58:00</td>\n",
       "      <td>2017-10-26 13:21:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>66</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>3 days 23:58:00</td>\n",
       "      <td>28 days 05:52:00</td>\n",
       "      <td>2017-10-30 13:19:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>67</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>28 days 05:52:00</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>2017-11-27 19:11:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>68</th>\n",
       "      <td>wikidatawiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>NaT</td>\n",
       "      <td>2017-11-27 19:11:00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         wiki_db has_ores has_rcfilters has_rcfilters_watchlist  \\\n",
       "63  wikidatawiki     True         False                   False   \n",
       "64  wikidatawiki     True          True                   False   \n",
       "65  wikidatawiki    False         False                   False   \n",
       "66  wikidatawiki     True          True                    True   \n",
       "67  wikidatawiki    False         False                   False   \n",
       "68  wikidatawiki     True          True                    True   \n",
       "\n",
       "    time_since_last_cutoff  time_till_next_cutoff                date  \n",
       "63                     NaT      321 days 08:41:00 2016-06-22 15:13:00  \n",
       "64       321 days 08:41:00      169 days 13:27:00 2017-05-09 23:54:00  \n",
       "65       169 days 13:27:00        3 days 23:58:00 2017-10-26 13:21:00  \n",
       "66         3 days 23:58:00       28 days 05:52:00 2017-10-30 13:19:00  \n",
       "67        28 days 05:52:00        0 days 00:00:00 2017-11-27 19:11:00  \n",
       "68         0 days 00:00:00                    NaT 2017-11-27 19:11:00  "
      ]
     },
     "execution_count": 263,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cutoffs.loc[cutoffs.wiki_db == 'wikidatawiki',select]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 264,
   "metadata": {},
   "outputs": [],
   "source": [
    "# wikidatawiki had an issue on 2017-10-30 and 2017-11-27 with the move to default on watchlist so we'll ignore that cutoff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 265,
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoffs = cutoffs.loc[~((cutoffs.wiki_db == 'wikidatawiki') & (cutoffs.date >= pd.to_datetime(\"2017-10-26 13:21:00\")))]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 266,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>wiki_db</th>\n",
       "      <th>has_ores</th>\n",
       "      <th>has_rcfilters</th>\n",
       "      <th>has_rcfilters_watchlist</th>\n",
       "      <th>time_since_last_cutoff</th>\n",
       "      <th>time_till_next_cutoff</th>\n",
       "      <th>date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>etwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>50 days 05:26:00</td>\n",
       "      <td>2017-03-20 18:28:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>etwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>50 days 05:26:00</td>\n",
       "      <td>308 days 17:24:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>22</th>\n",
       "      <td>frwiki</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>55 days 07:02:00</td>\n",
       "      <td>2017-04-11 11:09:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>23</th>\n",
       "      <td>frwiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>55 days 07:02:00</td>\n",
       "      <td>156 days 20:24:00</td>\n",
       "      <td>2017-06-05 18:11:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>27</th>\n",
       "      <td>hewiki</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaT</td>\n",
       "      <td>29 days 10:38:00</td>\n",
       "      <td>2017-04-10 13:16:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>28</th>\n",
       "      <td>hewiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>29 days 10:38:00</td>\n",
       "      <td>308 days 17:24:00</td>\n",
       "      <td>2017-05-09 23:54:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>33</th>\n",
       "      <td>kowiki</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>0 days 00:00:00</td>\n",
       "      <td>NaT</td>\n",
       "      <td>2019-03-04 16:27:00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   wiki_db has_ores has_rcfilters has_rcfilters_watchlist  \\\n",
       "12  etwiki     True         False                   False   \n",
       "13  etwiki     True          True                   False   \n",
       "22  frwiki    False         False                   False   \n",
       "23  frwiki     True          True                   False   \n",
       "27  hewiki     True         False                   False   \n",
       "28  hewiki     True          True                   False   \n",
       "33  kowiki     True          True                    True   \n",
       "\n",
       "    time_since_last_cutoff  time_till_next_cutoff                date  \n",
       "12                     NaT       50 days 05:26:00 2017-03-20 18:28:00  \n",
       "13        50 days 05:26:00      308 days 17:24:00 2017-05-09 23:54:00  \n",
       "22                     NaT       55 days 07:02:00 2017-04-11 11:09:00  \n",
       "23        55 days 07:02:00      156 days 20:24:00 2017-06-05 18:11:00  \n",
       "27                     NaT       29 days 10:38:00 2017-04-10 13:16:00  \n",
       "28        29 days 10:38:00      308 days 17:24:00 2017-05-09 23:54:00  \n",
       "33         0 days 00:00:00                    NaT 2019-03-04 16:27:00  "
      ]
     },
     "execution_count": 266,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cutoffs.loc[(cutoffs.time_since_last_cutoff<=pd.Timedelta(60,'D')) | (cutoffs.time_till_next_cutoff<=pd.Timedelta(60,'D')), select]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 267,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_cutoff = cutoffs.groupby(['wiki_db']).date.min()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 268,
   "metadata": {},
   "outputs": [],
   "source": [
    "# we'll look over the range of dates starting in June 2017 after thresholds in model info were released"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 270,
   "metadata": {},
   "outputs": [],
   "source": [
    "remember_dict[\"min.first.cutoff\"] = first_cutoff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 271,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_first_cutoff = datetime(year=2017,month=6,day=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 272,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_cutoff = first_cutoff.reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 273,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_cutoff.loc[(first_cutoff.date < min_first_cutoff),\"date\"] = min_first_cutoff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 274,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_cutoff = first_cutoff.reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 275,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_cutoff = cutoffs.merge(first_cutoff,how='right')[['wiki_db','date']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 276,
   "metadata": {},
   "outputs": [],
   "source": [
    "last_cutoff = cutoffs.groupby(['wiki_db']).date.max().reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 277,
   "metadata": {},
   "outputs": [],
   "source": [
    "last_cutoff = cutoffs.merge(last_cutoff,how='right')[['wiki_db','date']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 278,
   "metadata": {},
   "outputs": [],
   "source": [
    "fl_cutoff = last_cutoff.merge(first_cutoff,on='wiki_db',suffixes=['_last','_first'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 284,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd_period_length = pd.Timedelta(365,'d')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 285,
   "metadata": {},
   "outputs": [],
   "source": [
    "remember_dict['rdd_period_length'] = rdd_period_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 286,
   "metadata": {},
   "outputs": [],
   "source": [
    "fl_cutoff['period2_end'] = fl_cutoff.date_last + rdd_period_length\n",
    "fl_cutoff['period1_start'] = fl_cutoff.date_first - rdd_period_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 288,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = spark.read.table(\"wmf.mediawiki_history\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 289,
   "metadata": {},
   "outputs": [],
   "source": [
    "# take a stratified sample of edits in the cutoffs\n",
    "# stratify by wiki_db, is_newcomer, is_anon, is_reverted, and pre-ores and post-ores periods\n",
    "wmhist = spark.read.table(\"wmf.mediawiki_history\")\n",
    "\n",
    "wmhist = wmhist.filter(f.col(\"snapshot\") == \"2019-10\")\n",
    "# ok we're ready to fire up spark and make a stratified sample\n",
    "# we only need the latest snapshot\n",
    "\n",
    "wmhist = wmhist.filter((f.col(\"event_entity\") == \"revision\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 290,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 291,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import build_wmhist_step1, process_reverts, broadcast_match_comment, add_revert_types"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 292,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function spark_functions.broadcast_match_comment.<locals>.my_match_comment>"
      ]
     },
     "execution_count": 292,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "broadcast_match_comment(spark.sparkContext)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 293,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.filter(wmhist.page_namespace == 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 294,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "build_wmhist_step1() missing 1 required positional argument: 'remember_dict'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-294-fd5af7cc2200>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mwmhist\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mbuild_wmhist_step1\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mwmhist\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mTypeError\u001b[0m: build_wmhist_step1() missing 1 required positional argument: 'remember_dict'"
     ]
    }
   ],
   "source": [
    "wmhist = build_wmhist_step1(wmhist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts = process_reverts(wmhist,spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# select only the columns we need from reverts\n",
    "reverts = reverts.select(['wiki_db_l','revert_timestamp','reverted_revision_id',f.col('role_type').alias(\"revert_role_type\"),f.col('anon_new_established').alias('reverted_anon_new_established'),'is_damage','time_to_revert','revert_comment','revert_user_Nreverts_past_month','revert_user_text','revert_user_id'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for the time to revert analysis, we only want damaging ones, but for is_reverted we want all reverts\n",
    "wmhist = wmhist.join(reverts, on =[wmhist.wiki_db == reverts.wiki_db_l, wmhist.revision_id == reverts.reverted_revision_id],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist = wmhist.join(cutoffs, on=[wmhist.wiki_db==cutoffs.wiki_db_l, f.unix_timestamp(wmhist.event_timestamp) >= f.unix_timestamp(cutoffs.period_start), f.unix_timestamp(wmhist.event_timestamp) <= f.unix_timestamp(cutoffs.period_end)],how='right_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist = wmhist.withColumn(\"sec_to_cutoff\", (f.unix_timestamp(f.col(\"event_timestamp\")) - f.unix_timestamp(f.col(\"date\"))) / 1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist = add_revert_types(wmhist, comment_column='revert_comment')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# instead of taking a sample around every cutoff. We're going to just look 3 months before the earliest cutoff and 3 months after the latest one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fl_cutoff_df = spark.createDataFrame(fl_cutoff[['wiki_db','date_first','date_last','period1_start','period2_end']])\n",
    "fl_cutoff_df = fl_cutoff_df.withColumnRenamed('wiki_db','wiki_db_l')\n",
    "fl_cutoff_df = f.broadcast(fl_cutoff_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fl_cutoff_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "join_cond = (wmhist.wiki_db == fl_cutoff_df.wiki_db_l) & (f.unix_timestamp(wmhist.event_timestamp).between(f.unix_timestamp(fl_cutoff_df.period1_start),f.unix_timestamp(fl_cutoff_df.date_first)) | (f.unix_timestamp(wmhist.event_timestamp).between(f.unix_timestamp(fl_cutoff_df.date_last),f.unix_timestamp(fl_cutoff_df.period2_end))))\n",
    "\n",
    "                                                          "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist = wmhist.repartition(10000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.join(fl_cutoff_df,on=join_cond, how='inner')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ores_cutoff_cond = None\n",
    "# rcfilters_cutoff_cond = None\n",
    "# rcfilters_watchlist_cutoff_cond = None\n",
    "# for _, cutoff in cutoffs.iterrows():\n",
    "#     cond = ((wmhist.wiki_db == cutoff.wiki_db) & (f.unix_timestamp(wmhist.event_timestamp) >= cutoff.period_start.timestamp()) & (f.unix_timestamp(wmhist.event_timestamp) <= cutoff.period_end.timestamp()))\n",
    "            \n",
    "#     if ores_cutoff_cond is None:\n",
    "#         ores_cutoff_cond = (cond & f.lit(cutoff.ores_cutoff == True))\n",
    "#         rcfilters_cutoff_cond = (cond & f.lit(cutoff.rcfilters_cutoff == True))\n",
    "#         rcfilters_watchlist_cutoff_cond = (cond & f.lit(cutoff.watchlist_cutoff == True))\n",
    "#     else:\n",
    "#         ores_cutoff_cond = ores_cutoff_cond | (cond & f.lit(cutoff.ores_cutoff == True))\n",
    "#         rcfilters_cutoff_cond = rcfilters_cutoff_cond | (cond & f.lit(cutoff.rcfilters_cutoff == True))\n",
    "#         rcfilters_watchlist_cutoff_cond = rcfilters_watchlist_cutoff_cond | (cond & f.lit(cutoff.watchlist_cutoff == True))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# wmhist = wmhist.withColumn(\"cutoff_type\", f.when(ores_cutoff_cond,'has_ores').otherwise(f.when(rcfilters_cutoff_cond,'has_rcfilters').otherwise(f.when(rcfilters_watchlist_cutoff_cond,'has_rcfilters_watchlist').otherwise(None))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.withColumn(\"period\",f.when(f.unix_timestamp(wmhist.event_timestamp) <= f.unix_timestamp(wmhist.date_first),\"period1\").otherwise(\"period2\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.withColumn('strata',f.concat_ws('_',wmhist.wiki_db,wmhist.period,wmhist.revision_is_identity_reverted,wmhist.reverted_anon_new_established))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist.select(f.col(\"anon_new_established\"),f.col(\"reverted_anon_new_established\"),f.col(\"revision_id\"),f.col(\"revert_role_type\")).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.withColumn(\"reverted_in_24h\",(wmhist.revision_is_identity_reverted == True) & (wmhist.time_to_revert <= 60*60*24))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#wmhist = wmhist.repartition(1200,f.col(\"strata\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist_out = wmhist.select(['wiki_db','event_timestamp','page_id','page_title','user_id','user_text','event_user_isbot1','event_user_isbot2','revision_id','revision_is_identity_reverted','anon_new_established','event_user_is_newcomer','time_to_revert','reverted_in_24h','period','period1_start','period2_end','strata','date_first','date_last'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 295,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist_out = wmhist_out.repartition(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist_out.write.csv(\"/user/nathante/ores_bias_data/cutoff_revisions_2periods.csv\",mode='overwrite',compression=\"None\",header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = spark.read.csv(\"ores_bias_data/cutoff_revisions_2periods.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.repartition(500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get the proportion of observations in each strata\n",
    "strata_count = wmhist.groupby(f.col('strata')).count()\n",
    "#all_count = wmhist.count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "strata_count = strata_count.withColumn(\"fraction\", f.when( 20000 < f.col(\"count\"),20000/f.col(\"count\")).otherwise(1))\n",
    "strata_count = strata_count.withColumn(\"weight\",1/strata_count.fraction)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "remember_dict['strata_sample_size'] = 20000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "samp_design = strata_count.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#fractions = samp_design.locj['strata','fraction']\n",
    "fractions = {r.strata:r.fraction for r in samp_design}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = wmhist.sampleBy(\"strata\",fractions=fractions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = sample.join(strata_count,on='strata')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## TODO Write the table. check if it exists and get the sample revids from there.\n",
    "\n",
    "#sample.write.saveAsTable(\"nathante.cutoff_revisions_sample_simplestrata\",mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = sample.repartition(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample.write.csv(\"/user/nathante/ores_bias_data/cutoff_revisions_sample_2periods.csv\",mode='overwrite',compression='none',header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "strata_count.write.csv(\"/user/nathante/ores_bias_data/threshhold_strata_counts.csv\",mode='overwrite',compression=\"none\",header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------------------+\n",
      "|avg(CAST(revision_is_identity_reverted AS BIGINT))|\n",
      "+--------------------------------------------------+\n",
      "|                                0.5336424575374225|\n",
      "+--------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample.filter(sample.wiki_db=='enwiki').select(f.mean(f.col('revision_is_identity_reverted').cast(\"long\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = spark.sparkContext.getConf()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf.getAll()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits2 = edits.select(['wiki_db','event_timestamp','event_user_is_anonymous','event_user_is_anonymous','revision_id','revision_is_identity_reverted','rcfilters_cutoff','week','sec_to_cutoff'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.catalog.listDatabases()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ores_scores = spark.read.table(\"ores.revision_score_public\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ores_scores = ores_scores.filter((f.col(\"model\")==\"damaging\") & f.col(\"model_version\") == \"0.3.2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits = edits.join(ores_scores,on=[edits.wiki_db == ores_scores.wiki, edits.revision_id == ores_scores.rev_id])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits2 = edits2.withColumn(\"wikiweek\",f.concat_ws(' ', edits.wiki_db, f.date_format(edits.week,'MM-dd-yyyy')))\n",
    "by_wiki_week = edits.groupby('wikiweek')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "by_wiki_week = edits2.groupby(['wikiweek'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# take a sample stratified of N = 5000 by wiki and week\n",
    "\n",
    "samp_design = by_wiki_week.count()\n",
    "samp_design = samp_design.withColumn(\"fraction\", f.when( 5000 < f.col(\"count\"),5000/f.col(\"count\")).otherwise(1))\n",
    "samp_design = samp_design.withColumn(\"weight\",1/samp_design.fraction)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "samp_design"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fractions = samp_design.select(['wikiweek','fraction']).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fractions = {r.wikiweek:r.fraction for r in fractions}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fractions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = edits2.sampleBy(\"wikiweek\",fractions=fractions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pddf_sample = sample.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark - YARN - 12g",
   "language": "python",
   "name": "spark_yarn_pyspak_nathante_12g"
  },
  "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.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}