Twitter Task Stage 1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Start SparkNLP\n",
    "#\n",
    "from sparknlp.annotator import * \n",
    "from sparknlp.base import *\n",
    "from sparknlp.pretrained import PretrainedPipeline\n",
    "from pyspark.ml import Pipeline\n",
    "import sparknlp  \n",
    "from pyspark.sql.functions import *\n",
    "from nltk.corpus import stopwords\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml.feature import IDF, CountVectorizer, HashingTF\n",
    "\n",
    "import numpy as np\n",
    "import os\n",
    "\n",
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "spark = sparknlp.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "#load data into spark\n",
    "#using all the data is slow at my pc, so get just the first 100 000 tweets\n",
    "! head -n10000000 training_set_tweets.txt > tweets.txt\n",
    "tweets = spark.read.option(\"header\", \"false\").option(\"sep\", \"\\t\").csv(\"tweets.txt\")\n",
    "#\"/home/i/Downloads/twitter_cikm_2010/training_set_tweets.txt\") \n",
    "tweets = tweets.withColumnRenamed(\"_c0\", \"user_id\")\n",
    "tweets = tweets.withColumnRenamed(\"_c1\", \"tweet_id\")\n",
    "tweets = tweets.withColumnRenamed(\"_c2\", \"tweet\")\n",
    "tweets = tweets.withColumnRenamed(\"_c3\", \"dt\")\n",
    "tweets = tweets.where(\"tweet is NOT NULL\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lemma_antbnc download started this may take some time.\n",
      "Approximate size to download 907.6 KB\n",
      "[OK!]\n"
     ]
    }
   ],
   "source": [
    "docAsm = DocumentAssembler().setInputCol(\"tweet\").setOutputCol(\"document\")\n",
    "sentDet = SentenceDetector().setInputCols([\"document\"]).setOutputCol(\"sentences\")\n",
    "tokenizer = Tokenizer().setInputCols([\"sentences\"]).setOutputCol(\"tokens\")\n",
    "\n",
    "normalizer = Normalizer().setInputCols([\"tokens\"]).setOutputCol(\"normal\")\n",
    "lemmatizer = LemmatizerModel.pretrained().setInputCols([\"tokens\"]).setOutputCol(\"lemma\")\n",
    "    \n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Process tweets\n",
    "pipeline = Pipeline(stages=[docAsm, sentDet, tokenizer, normalizer, lemmatizer])\n",
    "m = pipeline.fit(tweets)\n",
    "tweets = m.transform(tweets)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "#twitter handles of top 100 most followed performers, https://www.theguardian.com/news/datablog/2013/apr/19/twitter-music-app-100-most-followed-musicians\n",
    "top_twitter_performers = spark.read.option(\"header\", \"true\").option(\"sep\", \",\").csv(\"twitter.top100.handles.csv\") \n",
    "#list of music artists, https://github.com/napsternxg/TwitterNER/tree/master/data/cleaned/custom_lexicons\n",
    "music_artist_names = spark.read.option(\"header\", \"true\").csv(\"music.artists.names.txt\") \n",
    "#select only full artist names - it seems the list contains many names which are look like ordinary words\n",
    "music_artist_names = music_artist_names.where(col(\"ArtistName\").contains(\" \"))\n",
    "#words used to detect wether a tweet has to do with a concert\n",
    "concert_detection_words = spark.read.option(\"header\", \"true\").csv(\"concert.keywords.txt\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "#filter tweets \n",
    "rel_tweets = tweets\\\n",
    "    .join(concert_detection_words, expr(\"array_contains(lemma.result, ConcertKeyword)\"), \"left\")\\\n",
    "    .where(\"(ConcertKeyword IS NOT NULL)\")\\\n",
    "    .join(music_artist_names, tweets.tweet.contains(music_artist_names.ArtistName), \"left\")\\\n",
    "    .join(top_twitter_performers, expr(\"array_contains(tokens.result, Handle)\"), \"left\")\\\n",
    "    .withColumn(\"Artist\", expr(\"CASE WHEN ArtistName IS NULL THEN TopPerformerName ELSE ArtistName END\"))\\\n",
    "    .where(\"(Artist IS NOT NULL)\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "rel_tweets.write.mode(\"overwrite\").save(\"rel.tweets.parquet\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.6.8 64-bit ('p3': venv)",
   "language": "python",
   "name": "python36864bitp3venv38f3a5947b564cb69cee0d670d158b95"
  },
  "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}