Build_wiki_weeks_simplified.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import timezone\n",
    "from datetime import datetime\n",
    "import pandas as pd\n",
    "import numpy as np\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 udf\n",
    "from dateutil import parser\n",
    "import pickle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "treated_wikis = pd.read_csv(\"data/ores_rcfilters_cutoffs.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# select canonical cutoffs\n",
    "wikis = set(treated_wikis.wiki_db)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ok we're ready to fire up spark and make a stratified sample\n",
    "wmhist = spark.read.table(\"wmf.mediawiki_history\")\n",
    "# we only need the latest snapshot\n",
    "wmhist = wmhist.filter(f.col(\"snapshot\") == \"2019-09\")\n",
    "wmhist = wmhist.filter((f.col(\"event_entity\") == \"revision\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "wmhist = wmhist.select(['wiki_db','event_timestamp','event_comment','revision_id','revision_parent_id','revision_text_bytes','revision_text_bytes_diff','revision_text_sha1','revision_is_identity_reverted','revision_first_identity_reverting_revision_id','revision_is_identity_revert','revision_tags','event_user_id','event_user_text','event_user_is_anonymous','event_user_creation_timestamp','event_user_first_edit_timestamp','event_user_revision_count','event_user_seconds_since_previous_revision','page_id','page_title_historical','page_title',                'page_namespace','page_is_redirect','page_is_deleted','page_revision_count','page_seconds_since_previous_revision',\"event_user_groups\",\"event_user_is_bot_by\",\"revision_deleted_parts\",'revision_deleted_parts_are_suppressed'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "wmhist = wmhist.filter(f.col(\"wiki_db\").isin(wikis))\n",
    "wmhist = wmhist.filter(f.col(\"page_namespace\") == 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for convenience we'll use the start of the week as the time of intervention and ignore the fact that it was actually mid-week in some cases\n",
    "wmhist = wmhist.withColumn(\"week\",f.date_trunc(\"week\",wmhist.event_timestamp))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "remember_dict = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_time = f.lit(\"2007-01-01\")\n",
    "wmhist = wmhist.filter(f.col('week') > min_time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "overwrite = False\n",
    "if os.path.exists(\"data/all_wiki_weeks.csv\") and not overwrite is True:\n",
    "    wiki_weeks = pd.read_csv(\"data/all_wiki_weeks.csv\",parse_dates=['week'])\n",
    "else:\n",
    "    wikis = wmhist.select(\"wiki_db\").distinct().toPandas()\n",
    "    weeks = wmhist.select(\"week\").distinct().toPandas()\n",
    "\n",
    "    wikis = wikis.assign(key=1)\n",
    "    weeks = weeks.assign(key=1)\n",
    "    wiki_weeks = wikis.merge(weeks,on='key')\n",
    "    wiki_weeks = wiki_weeks.drop('key',axis=1)\n",
    "    wiki_weeks.to_csv(\"data/all_wiki_weeks.csv\",index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import build_wmhist_step1\n",
    "wmhist, remember_dict = build_wmhist_step1(wmhist, remember_dict)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from spark_functions import process_reverts\n",
    "reverts, remember_dict = process_reverts(wmhist, spark, remember_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = spark.createDataFrame(wiki_weeks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# exclude reverts with ttr > 30 days = 60 seconds * 60 minutes / second * 24hours / day * 30 days           \n",
    "reverts = reverts.filter(f.col(\"time_to_revert\") <= 30*24*60*60)\n",
    "remember_dict['max_time_to_revert_days'] = 30                                                        \n",
    "\n",
    "# reverts = reverts.withColumn(\"med_ttr\", f.expr('percentile_approx(time_to_revert, 0.5,1)').over(Window.partitionBy(['wiki_db','week'])))                                                                       "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "reverts_by_week = reverts.groupBy(['wiki_db','week']).pivot(\"anon_new_established\",['anonymous','newcomer','established']).agg(f.count('reverted_revision_id').alias(\"N_reverted\"),\n",
    "                                                                                                                               f.mean(f.log('time_to_revert')).alias('geom_mean_ttr'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits_by_week = wmhist.filter(wmhist.role_type!='bot'). groupBy(['wiki_db','week']).pivot(\"anon_new_established\",['anonymous','newcomer','established']).agg(f.count(\"revision_id\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "edits_by_week = edits_by_week.withColumnRenamed(\"established\",\"established_N_edits\")\n",
    "edits_by_week = edits_by_week.withColumnRenamed(\"anonymous\",\"anonymous_N_edits\")\n",
    "edits_by_week = edits_by_week.withColumnRenamed(\"newcomer\",\"newcomer_N_edits\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(reverts_by_week,on=['wiki_db','week'],how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.join(edits_by_week,on=['wiki_db','week'], how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks = wiki_weeks.fillna(0,['anonymous_N_reverted','newcomer_N_reverted','established_N_reverted','anonymous_N_edits','newcomer_N_edits','established_N_edits'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out = wiki_weeks.repartition(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out = wiki_weeks_out.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "wiki_weeks_out.write.csv(\"/user/nathante/ores_bias_data/wiki_weeks_simplified.csv\",header=\"True\", compression=\"None\", mode=\"overwrite\")"
   ]
  },
  {
   "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
}