preprocessing_training.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark as ps\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import time\n",
    "\n",
    "from google.cloud import storage\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.ml.feature import Tokenizer, StopWordsRemover, Word2Vec, Word2VecModel\n",
    "\n",
    "from functools import reduce\n",
    "from itertools import islice\n",
    "\n",
    "\n",
    "from bz2 import BZ2File\n",
    "import xml.etree.ElementTree as etree\n",
    "\n",
    "from __future__ import print_function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Download ukrianian wikipedia dump (Datalab)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!wget https://dumps.wikimedia.org/ukwiki/latest/ukwiki-latest-pages-articles.xml.bz2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!bzip2 -dk ukwiki-20190701-pages-articles-multistream.xml.bz2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def strip_tag_name(t):\n",
    "    idx = k = t.rfind(\"}\")\n",
    "    if idx != -1:\n",
    "        return t[idx + 1:]\n",
    "    else:\n",
    "        return t\n",
    "    \n",
    "def read_wiki_dump(bz2_dump_path):\n",
    "    with BZ2File(bz2_dump_path) as xml_file:\n",
    "        for event, elem in etree.iterparse(xml_file, events=(\"start\", \"end\")):\n",
    "            tname = strip_tag_name(elem.tag)\n",
    "            if event == \"start\":\n",
    "                if tname == \"page\":\n",
    "                    title = \"\"\n",
    "                    redirect = \"\"\n",
    "                    ns = 0\n",
    "                    text = \"\"\n",
    "            else:\n",
    "                if tname == \"title\":\n",
    "                    title = elem.text\n",
    "                elif tname == \"redirect\":\n",
    "                    redirect = elem.attrib[\"title\"]\n",
    "                elif tname == \"ns\":\n",
    "                    ns = int(elem.text)\n",
    "                elif tname == \"text\":\n",
    "                    text = elem.text\n",
    "                elif tname == \"page\":\n",
    "                    yield title, redirect, ns, text\n",
    "                elem.clear()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Split large `.xml` file into small `.csv` chunks retaining only pages content."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "SLICE_SIZE = 2000000\n",
    "\n",
    "def is_article(title, redirect, ns, text):\n",
    "    return ns == 0 and len(redirect) == 0\n",
    "\n",
    "\n",
    "data = pd.DataFrame(columns=['Title', 'Text'])\n",
    "\n",
    "batch_size = 10000\n",
    "batch_id = 0\n",
    "count = 0\n",
    "\n",
    "for title, redirect, ns, text in islice(\n",
    "        filter(\n",
    "            lambda it: is_article(*it), \n",
    "            read_wiki_dump(\"ukwiki-20190701-pages-articles.xml.bz2\")\n",
    "        ), SLICE_SIZE):\n",
    "    data.loc[count, 'Title'] = title\n",
    "    data.loc[count, 'Text'] = text.replace(',', ' ').replace('\\n', ' ').replace('\\t', ' ').replace('\\r', ' ')\n",
    "    count += 1\n",
    "    \n",
    "    if count % batch_size == 0:\n",
    "        data.to_csv(\"ukwiki_\" + str(batch_id) + \".csv\")\n",
    "        data = pd.DataFrame(columns=['Title', 'Text'])\n",
    "        count = 0\n",
    "        batch_id += 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!gsutil cp ukwiki_*.csv gs://mmds/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Use word count to determine wikipedia service words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# words = spark.read.csv(\"preproc/preproc_0.csv\", header=True).select('vector_no_stopwords')\\\n",
    "#     .rdd.flatMap(lambda line: line[0].strip().split(\" \") if line[0] else str(line[0]))\n",
    "\n",
    "# wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda a,b:a + b).sortBy(lambda x: -x[1])\n",
    "\n",
    "# wordCounts.take(1000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Copy data from cloud storage to cluster hdfs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "19/07/20 12:10:58 INFO tools.OptionsParser: parseChunkSize: blocksperchunk false\n",
      "19/07/20 12:10:59 INFO tools.DistCp: Input Options: DistCpOptions{atomicCommit=false, syncFolder=false, deleteMissing=false, ignoreFailures=false, overwrite=false, append=false, useDiff=false, useRdiff=false, fromSnapshot=null, toSnapshot=null, skipCRC=false, blocking=true, numListstatusThreads=0, maxMaps=20, mapBandwidth=100, sslConfigurationFile='null', copyStrategy='uniformsize', preserveStatus=[], preserveRawXattrs=false, atomicWorkPath=null, logPath=null, sourceFileListing=null, sourcePaths=[gs:/ukwiki_*.csv], targetPath=hdfs:/, targetPathExists=true, filtersFile='null', blocksPerChunk=0, copyBufferSize=8192, verboseLog=false}\n",
      "19/07/20 12:11:00 INFO client.RMProxy: Connecting to ResourceManager at cluster-43cf-m/10.128.0.20:8032\n",
      "19/07/20 12:11:00 INFO client.AHSProxy: Connecting to Application History server at cluster-43cf-m/10.128.0.20:10200\n",
      "19/07/20 12:11:01 WARN gcs.GoogleHadoopFileSystem: GHFS.configureBuckets: Warning. No GCS bucket provided. Falling back on deprecated fs.gs.system.bucket.\n",
      "19/07/20 12:11:35 INFO tools.SimpleCopyListing: Paths (files+dirs) cnt = 91; dirCnt = 0\n",
      "19/07/20 12:11:35 INFO tools.SimpleCopyListing: Build file listing completed.\n",
      "19/07/20 12:11:35 INFO Configuration.deprecation: io.sort.mb is deprecated. Instead, use mapreduce.task.io.sort.mb\n",
      "19/07/20 12:11:35 INFO Configuration.deprecation: io.sort.factor is deprecated. Instead, use mapreduce.task.io.sort.factor\n",
      "19/07/20 12:11:35 INFO tools.DistCp: Number of paths in the copy list: 91\n",
      "19/07/20 12:11:35 INFO tools.DistCp: Number of paths in the copy list: 91\n",
      "19/07/20 12:11:35 INFO client.RMProxy: Connecting to ResourceManager at cluster-43cf-m/10.128.0.20:8032\n",
      "19/07/20 12:11:35 INFO client.AHSProxy: Connecting to Application History server at cluster-43cf-m/10.128.0.20:10200\n",
      "19/07/20 12:11:35 INFO mapreduce.JobSubmitter: number of splits:23\n",
      "19/07/20 12:11:35 INFO Configuration.deprecation: yarn.resourcemanager.system-metrics-publisher.enabled is deprecated. Instead, use yarn.system-metrics-publisher.enabled\n",
      "19/07/20 12:11:35 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1563606014871_0007\n",
      "19/07/20 12:11:35 INFO impl.YarnClientImpl: Submitted application application_1563606014871_0007\n",
      "19/07/20 12:11:35 INFO mapreduce.Job: The url to track the job: http://cluster-43cf-m:8088/proxy/application_1563606014871_0007/\n",
      "19/07/20 12:11:35 INFO tools.DistCp: DistCp job-id: job_1563606014871_0007\n",
      "19/07/20 12:11:35 INFO mapreduce.Job: Running job: job_1563606014871_0007\n",
      "19/07/20 12:11:44 INFO mapreduce.Job: Job job_1563606014871_0007 running in uber mode : false\n",
      "19/07/20 12:11:44 INFO mapreduce.Job:  map 0% reduce 0%\n",
      "19/07/20 12:11:52 INFO mapreduce.Job:  map 4% reduce 0%\n",
      "19/07/20 12:11:58 INFO mapreduce.Job:  map 9% reduce 0%\n",
      "19/07/20 12:12:04 INFO mapreduce.Job:  map 17% reduce 0%\n",
      "19/07/20 12:12:09 INFO mapreduce.Job:  map 22% reduce 0%\n",
      "19/07/20 12:12:10 INFO mapreduce.Job:  map 30% reduce 0%\n",
      "19/07/20 12:12:16 INFO mapreduce.Job:  map 39% reduce 0%\n",
      "19/07/20 12:12:17 INFO mapreduce.Job:  map 43% reduce 0%\n",
      "19/07/20 12:12:22 INFO mapreduce.Job:  map 48% reduce 0%\n",
      "19/07/20 12:12:28 INFO mapreduce.Job:  map 52% reduce 0%\n",
      "19/07/20 12:12:40 INFO mapreduce.Job:  map 64% reduce 0%\n",
      "19/07/20 12:12:41 INFO mapreduce.Job:  map 70% reduce 0%\n",
      "19/07/20 12:12:42 INFO mapreduce.Job:  map 73% reduce 0%\n",
      "19/07/20 12:12:43 INFO mapreduce.Job:  map 74% reduce 0%\n",
      "19/07/20 12:12:44 INFO mapreduce.Job:  map 83% reduce 0%\n",
      "19/07/20 12:12:46 INFO mapreduce.Job:  map 87% reduce 0%\n",
      "19/07/20 12:12:47 INFO mapreduce.Job:  map 91% reduce 0%\n",
      "19/07/20 12:12:48 INFO mapreduce.Job:  map 100% reduce 0%\n",
      "19/07/20 12:12:49 INFO mapreduce.Job: Job job_1563606014871_0007 completed successfully\n",
      "19/07/20 12:12:49 INFO mapreduce.Job: Counters: 37\n",
      "\tFile System Counters\n",
      "\t\tFILE: Number of bytes read=0\n",
      "\t\tFILE: Number of bytes written=4869596\n",
      "\t\tFILE: Number of read operations=0\n",
      "\t\tFILE: Number of large read operations=0\n",
      "\t\tFILE: Number of write operations=0\n",
      "\t\tGS: Number of bytes read=0\n",
      "\t\tGS: Number of bytes written=0\n",
      "\t\tGS: Number of read operations=0\n",
      "\t\tGS: Number of large read operations=0\n",
      "\t\tGS: Number of write operations=0\n",
      "\t\tHDFS: Number of bytes read=20917\n",
      "\t\tHDFS: Number of bytes written=2720\n",
      "\t\tHDFS: Number of read operations=298\n",
      "\t\tHDFS: Number of large read operations=0\n",
      "\t\tHDFS: Number of write operations=69\n",
      "\tJob Counters \n",
      "\t\tLaunched map tasks=23\n",
      "\t\tOther local map tasks=23\n",
      "\t\tTotal time spent by all maps in occupied slots (ms)=490725\n",
      "\t\tTotal time spent by all reduces in occupied slots (ms)=0\n",
      "\t\tTotal time spent by all map tasks (ms)=490725\n",
      "\t\tTotal vcore-milliseconds taken by all map tasks=490725\n",
      "\t\tTotal megabyte-milliseconds taken by all map tasks=502502400\n",
      "\tMap-Reduce Framework\n",
      "\t\tMap input records=91\n",
      "\t\tMap output records=91\n",
      "\t\tInput split bytes=3082\n",
      "\t\tSpilled Records=0\n",
      "\t\tFailed Shuffles=0\n",
      "\t\tMerged Map outputs=0\n",
      "\t\tGC time elapsed (ms)=9893\n",
      "\t\tCPU time spent (ms)=56040\n",
      "\t\tPhysical memory (bytes) snapshot=6168510464\n",
      "\t\tVirtual memory (bytes) snapshot=58976169984\n",
      "\t\tTotal committed heap usage (bytes)=5669650432\n",
      "\tFile Input Format Counters \n",
      "\t\tBytes Read=17835\n",
      "\tFile Output Format Counters \n",
      "\t\tBytes Written=2720\n",
      "\tDistCp Counters\n",
      "\t\tBytes Skipped=6333247355\n",
      "\t\tFiles Skipped=91\n"
     ]
    }
   ],
   "source": [
    "!hadoop distcp gs:///ukwiki_*.csv hdfs://"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Init spark and storage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.master(\"local\").appName(\"WikiParse\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Download stop words and service words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sw_df = spark.read.csv(\"gs:///stop_words.csv\").toPandas()\n",
    "sr_df = spark.read.csv(\"gs:///service_words.csv\").toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stop_words = sw_df['_c0'].tolist()\n",
    "service_words = sr_df['_c0'].tolist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Read all of the training data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_batch(offset, limit):\n",
    "    fractions = []\n",
    "    for i in range(offset, limit):\n",
    "        print(\"Downloading fraction number {}...\".format(i))\n",
    "        df = spark.read.format(\"csv\") \\\n",
    "                        .option(\"header\", \"true\") \\\n",
    "                        .option(\"encoding\", \"UTF-8\") \\\n",
    "                        .load(\"hdfs:///ukwiki_\" + str(i) + \".csv\")\n",
    "        fractions.append(df)\n",
    "    return fractions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "scrolled": false
   },
   "source": [
    "# Create training pipeline for word2vec model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Pipeline:\n",
    "    def __init__(self, df, stop_words, service_words):\n",
    "        self.stop_words = stop_words\n",
    "        self.service_words = service_words\n",
    "        self.df = df\n",
    "        self.vector_df = None\n",
    "    \n",
    "    def fit(self, sample=0):\n",
    "        word2vec = Word2Vec(vectorSize=100, seed=42, inputCol='text', outputCol='model')\n",
    "        \n",
    "        if sample == 0:\n",
    "            return word2Vec.fit(self.vector_df)\n",
    "        \n",
    "        part = self.vector_df.take(sample)\n",
    "        \n",
    "        model = word2vec.fit(spark.createDataFrame(part, schema=self.vector_df.schema))\n",
    "        return model\n",
    "    \n",
    "    def preprocess(self):\n",
    "        # clean data\n",
    "        df_trip = self.df.select(['Title', 'Text'])\\\n",
    "            .withColumn('Text', regexp_replace('Text', '[§»«·&\\~.a-zA-Z^=\\-\\\"<>!?:;{}()\\[\\]/|%0-9\\\\\\+\\*#_]+', ' '))\\\n",
    "            .withColumn('Text', regexp_replace('Text', '\\'{3}', ' '))\\\n",
    "            .withColumn('Text', regexp_replace('Text', '[—−]', ' '))\\\n",
    "            .withColumn('Text', regexp_replace('Text', '[^а-яА-ЯіІіІєЄҐґїЇ\\s]', ''))\\\n",
    "            .withColumn('Text', regexp_replace('Text', '\\s+', ' '))\\\n",
    "            .select([trim(lower(col('Title'))).alias('Title'), trim(lower(col('Text'))).alias('Text')])\n",
    "        \n",
    "        # tokenize data\n",
    "        tokenizer = Tokenizer(inputCol=\"Text\", outputCol=\"Vector\")\n",
    "        self.vector_df = tokenizer.transform(df_trip).select(\"vector\")\n",
    "        self.vector_df.show(5)\n",
    "        \n",
    "        # remove stop words\n",
    "        self.vector_df = self.__remove_stop_words(\"vector\", \"vector_no_stopwords\", self.stop_words)\n",
    "        \n",
    "        # remove service words\n",
    "        self.vector_df = self.__remove_stop_words(\"vector_no_stopwords\", \"text\", self.service_words)\n",
    "        \n",
    "    def __remove_stop_words(self, in_col, out_col, words_list):\n",
    "        remover = StopWordsRemover(inputCol=in_col, outputCol=out_col, stopWords=words_list)\n",
    "        stopwords = remover.getStopWords() \n",
    "        \n",
    "        vector_no_stopw_df = remover.transform(self.vector_df).select(out_col)\n",
    "        vector_no_stopw_df.show(5)\n",
    "        \n",
    "        return vector_no_stopw_df\n",
    "    \n",
    "    @staticmethod\n",
    "    def from_batch(batch, stop_words, service_words):\n",
    "        return Pipeline(batch, stop_words, service_words)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Training word2vec model over fraction of dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# # example for 10 articles\n",
    "\n",
    "# n = 10\n",
    "# print(\"Take {} rows\".format(n))\n",
    "# model = p.fit(sample=n)\n",
    "# print(\"Saving word2vec of {} rows\".format(n))\n",
    "# model.save(\"gs:///w2v/word2vec_{}\".format(n))\n",
    "\n",
    "# x = Word2VecModel.load(\"gs:///w2v/word2vec_{}\".format(n))\n",
    "# x.getVectors()\n",
    "\n",
    "# x.findSynonyms('дата', 5).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "start = 1\n",
    "end = 91\n",
    "\n",
    "for i in range(start, end, 10):\n",
    "    fractions = reduce(lambda x, y: x.union(y), read_batch(0, i))\n",
    "    p = Pipeline.from_batch(fractions, stop_words, service_words)\n",
    "    p.preprocess()\n",
    "    part = i*10000\n",
    "    print(\"Take {} rows\".format(part))\n",
    "    model = p.fit(sample=part)\n",
    "    print(\"Saving word2vec of {} rows\".format(part))\n",
    "    model.save(\"gs:///w2v/word2vec_{}\".format(i))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}