analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "from matplotlib.dates import DateFormatter\n",
    "import datetime\n",
    "import itertools\n",
    "from bs4 import BeautifulSoup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages com.datastax.spark:spark-cassandra-connector_2.11:2.3.0 --conf spark.cassandra.connection.host=127.0.0.1 pyspark-shell'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext(\"local\", \"analysis\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_df = sqlContext.read\\\n",
    "    .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "    .options(table='posts', keyspace='pw_project')\\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(table_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "table_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_accepted(x):\n",
    "    if x != -1:\n",
    "        return 1\n",
    "    else:\n",
    "        return 0\n",
    "udf_is_accepted = udf(is_accepted, IntegerType())\n",
    "\n",
    "table_df = table_df.withColumn(\"is_accepted\", udf_is_accepted(\"accepted_answer_id\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_histogram(var, min_ = None, max_ = None, bins = 30, label_space = 1):\n",
    "    x = table_df.select(var)\n",
    "    if min_ is not None:\n",
    "        x = x.where(x[var] > min_)\n",
    "    if max_ is not None:\n",
    "        x = x.where(x[var] < max_)\n",
    "    binSides, binCounts = x.rdd.flatMap(lambda x: x).histogram(bins)\n",
    "    binSides = [round(x,2) for x in binSides]\n",
    "    print(list(zip(binSides, binCounts)))\n",
    "    N = len(binCounts)\n",
    "    index = np.arange(N)\n",
    "    fig, ax = plt.subplots()\n",
    "    rects1 = ax.bar(index+0.5, binCounts, 1)\n",
    "    ax.set_xticks(np.arange(N+1))\n",
    "    ax.set_xticklabels([val if key%label_space == 0 else \"\" for key, val in enumerate(binSides)])\n",
    "    plt.show()\n",
    "\n",
    "def plot_var_sum_per_date(var):\n",
    "    var_sum_per_date = table_df.select([\"creation_date\", var]).rdd\\\n",
    "        .map(lambda x: (datetime.date(x[0].year, x[0].month, 1), x[1]))\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .sortBy(lambda x: x[0])\\\n",
    "        .collect()\n",
    "    print(var_sum_per_date)\n",
    "    x, y = zip(*var_sum_per_date)\n",
    "    fig, ax = plt.subplots()\n",
    "    ax.plot(x, y)\n",
    "    ax.xaxis.set_major_formatter(DateFormatter(\"%m/%y\"))\n",
    "    plt.show()\n",
    "\n",
    "def plot_var_avg_per_date(var):\n",
    "    tmp_tuple = (0,0)\n",
    "    var_avg_per_date = table_df.select([\"creation_date\", var]).rdd\\\n",
    "        .map(lambda x: (datetime.date(x[0].year, x[0].month, 1), x[1]))\\\n",
    "        .aggregateByKey(tmp_tuple, lambda a,b: (a[0] + b, a[1] + 1), lambda a,b: (a[0] + b[0], a[1] + b[1]))\\\n",
    "        .mapValues(lambda x: x[0]/x[1])\\\n",
    "        .sortBy(lambda x: x[0])\\\n",
    "        .collect()\n",
    "    print(var_avg_per_date)\n",
    "    x, y = zip(*var_avg_per_date)\n",
    "    fig, ax = plt.subplots()\n",
    "    ax.plot(x, y)\n",
    "    ax.xaxis.set_major_formatter(DateFormatter(\"%m/%y\"))\n",
    "    plt.show()\n",
    "\n",
    "def summarize_cont(var):\n",
    "    print(table_df.describe([var]).show())\n",
    "    print(table_df.where(table_df[var] > 0).describe([var]).show())\n",
    "    plot_histogram(var)\n",
    "    plot_var_sum_per_date(var)\n",
    "    plot_var_avg_per_date(var)\n",
    "\n",
    "def summarize_id(var):\n",
    "    print(table_df.where(table_df[var] > 0).count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# count\n",
    "# we can use post_type_id\n",
    "plot_var_sum_per_date(\"post_type_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# score\n",
    "summarize_cont(\"score\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_histogram(\"score\", -3, 16, 17, 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# view count\n",
    "summarize_cont(\"view_count\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "plot_histogram(\"view_count\", 0, 1000, 95, 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# answer count statistics\n",
    "summarize_cont(\"answer_count\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_histogram(\"answer_count\", -1, 18, 17, 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# comment count statistics\n",
    "summarize_cont(\"comment_count\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_histogram(\"comment_count\", -1, 40, 39, 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# favorite count statistics\n",
    "summarize_cont(\"favorite_count\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_histogram(\"favorite_count\", -1, 11, 10, 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_df.describe(['is_accepted']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_df.select('tags').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# najpopularniejsze tagi\n",
    "# count\n",
    "count_per_tag = table_df.select('tags').rdd\\\n",
    "    .flatMap(lambda x: x)\\\n",
    "    .flatMap(lambda x: tuple(x))\\\n",
    "    .map(lambda word: (word, 1))\\\n",
    "    .reduceByKey(lambda a, b: a + b)\\\n",
    "    .takeOrdered(50, key=lambda x: -x[1])\n",
    "count_per_tag_list = list(zip(*count_per_tag))[0]\n",
    "count_per_tag"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_var_per_tag(df):\n",
    "    xlabels, values = zip(*df[:20])\n",
    "    xlabels = [i[:15] + '...' if len(i) > 15 else i for i in xlabels]\n",
    "    N = len(xlabels)\n",
    "    index = np.arange(N)\n",
    "    fig, ax = plt.subplots()\n",
    "    rects1 = ax.bar(index, values, 1)\n",
    "    ax.set_xticks(np.arange(N+1))\n",
    "    ax.set_xticklabels(xlabels, rotation = 90)\n",
    "    plt.show()\n",
    "\n",
    "plot_var_per_tag(count_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def var_per_tag(var):\n",
    "    res = table_df.select([var, 'tags']).rdd\\\n",
    "        .flatMapValues(lambda x: x)\\\n",
    "        .map(lambda x: (x[1], x[0]))\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .takeOrdered(50, key=lambda x: -x[1])\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def var_avg_per_tag(var, tag_list):\n",
    "    tmp_tuple = (0,0)\n",
    "    res = table_df.select([var, 'tags']).rdd\\\n",
    "        .flatMapValues(lambda x: x)\\\n",
    "        .map(lambda x: (x[1], x[0]))\\\n",
    "        .filter(lambda x: x[0] in tag_list)\\\n",
    "        .aggregateByKey(tmp_tuple, lambda a,b: (a[0] + b, a[1] + 1), lambda a,b: (a[0] + b[0], a[1] + b[1]))\\\n",
    "        .mapValues(lambda x: x[0]/x[1])\\\n",
    "        .takeOrdered(50, key=lambda x: -x[1])\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# top tags\n",
    "# score\n",
    "score_per_tag = var_per_tag('score')\n",
    "score_per_tag_list = list(zip(*score_per_tag))[0]\n",
    "print(score_per_tag)\n",
    "plot_var_per_tag(score_per_tag)\n",
    "score_avg_per_tag = var_avg_per_tag('score', score_per_tag_list[:100])\n",
    "print(score_avg_per_tag)\n",
    "plot_var_per_tag(score_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# views\n",
    "view_per_tag = var_per_tag('view_count')\n",
    "view_per_tag_list = list(zip(*view_per_tag))[0]\n",
    "print(view_per_tag)\n",
    "plot_var_per_tag(view_per_tag)\n",
    "view_avg_per_tag = var_avg_per_tag('view_count', view_per_tag_list[:100])\n",
    "print(view_avg_per_tag)\n",
    "plot_var_per_tag(view_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# answer\n",
    "answer_per_tag = var_per_tag('answer_count')\n",
    "answer_per_tag_list = list(zip(*answer_per_tag))[0]\n",
    "print(answer_per_tag)\n",
    "plot_var_per_tag(answer_per_tag)\n",
    "answer_avg_per_tag = var_avg_per_tag('answer_count', answer_per_tag_list[:100])\n",
    "print(answer_avg_per_tag)\n",
    "plot_var_per_tag(answer_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# comments\n",
    "comment_per_tag = var_per_tag('comment_count')\n",
    "comment_per_tag_list = list(zip(*comment_per_tag))[0]\n",
    "print(comment_per_tag)\n",
    "plot_var_per_tag(comment_per_tag)\n",
    "comment_avg_per_tag = var_avg_per_tag('comment_count', comment_per_tag_list[:100])\n",
    "print(comment_avg_per_tag)\n",
    "plot_var_per_tag(comment_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# favorites\n",
    "favorite_per_tag = var_per_tag('favorite_count')\n",
    "favorite_per_tag_list = list(zip(*favorite_per_tag))[0]\n",
    "print(favorite_per_tag)\n",
    "plot_var_per_tag(favorite_per_tag)\n",
    "favorite_avg_per_tag = var_avg_per_tag('favorite_count', favorite_per_tag_list[:100])\n",
    "print(favorite_avg_per_tag)\n",
    "plot_var_per_tag(favorite_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# is accepted\n",
    "accepted_per_tag = var_per_tag('is_accepted')\n",
    "accepted_per_tag_list = list(zip(*accepted_per_tag))[0]\n",
    "print(accepted_per_tag)\n",
    "plot_var_per_tag(accepted_per_tag)\n",
    "accepted_avg_per_tag = var_avg_per_tag('is_accepted', accepted_per_tag_list[:100])\n",
    "print(accepted_avg_per_tag)\n",
    "plot_var_per_tag(accepted_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# popularity of tags within given month\n",
    "# filtered to top tags for a given criterion\n",
    "count_sum_per_tag_month = table_df.select(['creation_date', 'tags']).rdd\\\n",
    "    .flatMapValues(lambda x: x)\\\n",
    "    .filter(lambda x: x[1] in count_per_tag_list[0:9])\\\n",
    "    .map(lambda x: ((x[1], datetime.date(x[0].year, x[0].month, 1)), 1))\\\n",
    "    .reduceByKey(lambda a, b: a + b)\\\n",
    "    .map(lambda x: (x[0][0], x[0][1], x[1]))\\\n",
    "    .sortBy(lambda x: (x[0], x[1]))\\\n",
    "    .collect()\n",
    "\n",
    "count_sum_per_tag_month"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def cum_sum(input):\n",
    "    input_sorted = sorted(input, key=lambda x: x[0], reverse=False)\n",
    "    res_time = [i[0] for i in input_sorted]\n",
    "    res_cumsum = np.cumsum([i[1] for i in input_sorted]).tolist()\n",
    "    res = list(zip(res_time, res_cumsum))\n",
    "    return res\n",
    "\n",
    "count_cumsum_per_tag_month = table_df.select(['creation_date', 'tags']).rdd\\\n",
    "    .flatMapValues(lambda x: x)\\\n",
    "    .filter(lambda x: x[1] in count_per_tag_list[0:9])\\\n",
    "    .map(lambda x: ((x[1], datetime.date(x[0].year, x[0].month, 1)), 1))\\\n",
    "    .reduceByKey(lambda a, b: a + b)\\\n",
    "    .map(lambda x: (x[0][0], (x[0][1], x[1])))\\\n",
    "    .groupByKey()\\\n",
    "    .flatMapValues(lambda x: cum_sum(x))\\\n",
    "    .map(lambda x: (x[0], x[1][0], x[1][1]))\\\n",
    "    .collect()\n",
    "\n",
    "count_cumsum_per_tag_month"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def plot_var_per_tag_month(df, tag_list):\n",
    "    tag_list = tag_list[:9]\n",
    "    \n",
    "    palette = plt.get_cmap('Set1')\n",
    "    fig = plt.figure(figsize=(10, 10))\n",
    "\n",
    "    num=0\n",
    "    for tag in tag_list:\n",
    "        num+=1\n",
    "\n",
    "        ax = fig.add_subplot(3, 3, num)\n",
    "\n",
    "        for v in [i for i in tag_list if i != tag]:\n",
    "            x = [i[1] for i in df if i[0]==v]\n",
    "            y = [i[2] for i in df if i[0]==v]\n",
    "            ax.plot(x, y, marker='', color='grey', linewidth=0.6, alpha=0.3)\n",
    "\n",
    "        x = [i[1] for i in df if i[0]==tag]\n",
    "        y = [i[2] for i in df if i[0]==tag]\n",
    "        ax.plot(x, y, marker='', color=palette(num), linewidth=2.4, alpha=0.9, label=tag)\n",
    "\n",
    "        if num in range(7) :\n",
    "            ax.tick_params(labelbottom='off')\n",
    "        if num not in [1,4,7] :\n",
    "            ax.tick_params(labelleft='off')\n",
    "\n",
    "        ax.xaxis.set_major_formatter(DateFormatter(\"%m/%y\"))\n",
    "\n",
    "        ax.set_title(tag, loc='left', fontsize=12, fontweight=0, color=palette(num))\n",
    "\n",
    "    plt.show()\n",
    "\n",
    "\n",
    "plot_var_per_tag_month(count_sum_per_tag_month, count_per_tag_list)\n",
    "plot_var_per_tag_month(count_cumsum_per_tag_month, count_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def var_sum_per_tag_month(var, tag_list):\n",
    "    tag_list = tag_list[:9]\n",
    "    res = table_df.select(['creation_date', var, 'tags']).rdd\\\n",
    "        .map(lambda x: ((x[0], x[1]), x[2]))\\\n",
    "        .flatMapValues(lambda x: x)\\\n",
    "        .filter(lambda x: x[1] in tag_list)\\\n",
    "        .map(lambda x: ((x[1], datetime.date(x[0][0].year, x[0][0].month, 1)), x[0][1]))\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .map(lambda x: (x[0][0], x[0][1], x[1]))\\\n",
    "        .sortBy(lambda x: (x[0], x[1]))\\\n",
    "        .collect()\n",
    "    return res\n",
    "\n",
    "\n",
    "def var_cumsum_per_tag_month(var, tag_list):\n",
    "    tag_list = tag_list[:9]\n",
    "    res = table_df.select(['creation_date', var, 'tags']).rdd\\\n",
    "        .map(lambda x: ((x[0], x[1]), x[2]))\\\n",
    "        .flatMapValues(lambda x: x)\\\n",
    "        .filter(lambda x: x[1] in tag_list)\\\n",
    "        .map(lambda x: ((x[1], datetime.date(x[0][0].year, x[0][0].month, 1)), x[0][1]))\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .map(lambda x: (x[0][0], (x[0][1], x[1])))\\\n",
    "        .groupByKey()\\\n",
    "        .flatMapValues(lambda x: cum_sum(x))\\\n",
    "        .map(lambda x: (x[0], x[1][0], x[1][1]))\\\n",
    "        .collect()\n",
    "    return res\n",
    "\n",
    "score_sum_per_tag_month = var_sum_per_tag_month('score', score_per_tag_list)\n",
    "print(score_sum_per_tag_month)\n",
    "plot_var_per_tag_month(score_sum_per_tag_month, score_per_tag_list)\n",
    "score_cumsum_per_tag_month = var_cumsum_per_tag_month('score', score_per_tag_list)\n",
    "print(score_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(score_cumsum_per_tag_month, score_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "view_sum_per_tag_month = var_sum_per_tag_month('view_count', view_per_tag_list)\n",
    "print(view_sum_per_tag_month)\n",
    "plot_var_per_tag_month(view_sum_per_tag_month, view_per_tag_list)\n",
    "view_cumsum_per_tag_month = var_cumsum_per_tag_month('view_count', view_per_tag_list)\n",
    "print(view_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(view_cumsum_per_tag_month, view_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "answer_sum_per_tag_month = var_sum_per_tag_month('answer_count', answer_per_tag_list)\n",
    "print(answer_sum_per_tag_month)\n",
    "plot_var_per_tag_month(answer_sum_per_tag_month, answer_per_tag_list)\n",
    "answer_cumsum_per_tag_month = var_cumsum_per_tag_month('answer_count', answer_per_tag_list)\n",
    "print(answer_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(answer_cumsum_per_tag_month, answer_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "comment_sum_per_tag_month = var_sum_per_tag_month('comment_count', comment_per_tag_list)\n",
    "print(comment_sum_per_tag_month)\n",
    "plot_var_per_tag_month(comment_sum_per_tag_month, comment_per_tag_list)\n",
    "comment_cumsum_per_tag_month = var_cumsum_per_tag_month('comment_count', comment_per_tag_list)\n",
    "print(comment_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(comment_cumsum_per_tag_month, comment_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "favorite_sum_per_tag_month = var_sum_per_tag_month('favorite_count', favorite_per_tag_list)\n",
    "print(favorite_sum_per_tag_month)\n",
    "plot_var_per_tag_month(favorite_sum_per_tag_month, favorite_per_tag_list)\n",
    "favorite_cumsum_per_tag_month = var_cumsum_per_tag_month('favorite_count', favorite_per_tag_list)\n",
    "print(favorite_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(favorite_cumsum_per_tag_month, favorite_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "accepted_sum_per_tag_month = var_sum_per_tag_month('is_accepted', accepted_per_tag_list)\n",
    "print(accepted_sum_per_tag_month)\n",
    "plot_var_per_tag_month(accepted_sum_per_tag_month, accepted_per_tag_list)\n",
    "accepted_cumsum_per_tag_month = var_cumsum_per_tag_month('is_accepted', accepted_per_tag_list)\n",
    "print(accepted_cumsum_per_tag_month)\n",
    "plot_var_per_tag_month(accepted_cumsum_per_tag_month, accepted_per_tag_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "count_per_tag_comb = table_df.select('tags').rdd\\\n",
    "    .flatMap(lambda x: x)\\\n",
    "    .map(lambda x: list(itertools.combinations_with_replacement(x, 2)))\\\n",
    "    .flatMap(lambda x: x)\\\n",
    "    .filter(lambda x: x[0] != x[1])\\\n",
    "    .map(lambda x: (x, 1))\\\n",
    "    .reduceByKey(lambda a, b: a + b)\\\n",
    "    .takeOrdered(999, key=lambda x: -x[1])\n",
    "\n",
    "count_per_tag_comb[:50]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def plot_var_per_tag_comb(df, limit = 100):\n",
    "    df2 = df[:limit] + [((i[0][1], i[0][0]), i[1]) for i in df[:limit]]\n",
    "    rows, row_pos = np.unique([i[0][0] for i in df2], return_inverse=True)\n",
    "    cols, col_pos = np.unique([i[0][1] for i in df2], return_inverse=True)\n",
    "    pivoted_arr = np.zeros((len(rows), len(cols)))\n",
    "    pivoted_arr[row_pos, col_pos] = [i[1] for i in df2]\n",
    "\n",
    "    fig, ax = plt.subplots(figsize=(10, 10))\n",
    "    im = ax.imshow(pivoted_arr)\n",
    "\n",
    "    ax.set_xticks(np.arange(len(rows)))\n",
    "    ax.set_yticks(np.arange(len(cols)))\n",
    "    ax.set_xticklabels(rows)\n",
    "    ax.set_yticklabels(cols)\n",
    "\n",
    "    plt.setp(ax.get_xticklabels(), rotation=45, ha=\"right\", rotation_mode=\"anchor\")\n",
    "    fig.tight_layout()\n",
    "    plt.show()\n",
    "\n",
    "plot_var_per_tag_comb(count_per_tag_comb, 100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def var_per_tag_comb(var):\n",
    "    res = table_df.select([var, 'tags']).rdd\\\n",
    "        .mapValues(lambda x: x)\\\n",
    "        .map(lambda x: (x[0], list(itertools.combinations_with_replacement(x[1], 2))))\\\n",
    "        .flatMapValues(lambda x: x)\\\n",
    "        .map(lambda x: (x[1], x[0]))\\\n",
    "        .filter(lambda x: x[0][0] != x[0][1])\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .takeOrdered(50, key=lambda x: -x[1])\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "score_per_tag_comb = var_per_tag_comb('score')\n",
    "print(score_per_tag_comb)\n",
    "plot_var_per_tag_comb(score_per_tag_comb, 100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "view_per_tag_comb = var_per_tag_comb('view_count')\n",
    "print(view_per_tag_comb)\n",
    "plot_var_per_tag_comb(view_per_tag_comb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "comment_per_tag_comb = var_per_tag_comb('comment_count')\n",
    "print(comment_per_tag_comb)\n",
    "plot_var_per_tag_comb(comment_per_tag_comb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "favorite_per_tag_comb = var_per_tag_comb('favorite_count')\n",
    "print(favorite_per_tag_comb)\n",
    "plot_var_per_tag_comb(favorite_per_tag_comb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# count of words\n",
    "def count_words(x):\n",
    "    return len(str(x).split(\" \"))\n",
    "udf_count_words = udf(count_words, IntegerType())\n",
    "\n",
    "table_df = table_df.withColumn(\"count_words\", udf_count_words(\"body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "words_per_tag = var_per_tag('count_words')\n",
    "words_per_tag_list = list(zip(*words_per_tag))[0]\n",
    "print(words_per_tag)\n",
    "plot_var_per_tag(words_per_tag)\n",
    "words_avg_per_tag = var_avg_per_tag('count_words', words_per_tag_list[:100])\n",
    "print(words_avg_per_tag)\n",
    "plot_var_per_tag(words_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# count unique words\n",
    "def count_unique_words(x):\n",
    "    return len(set(str(x).lower().split(\" \")))\n",
    "udf_count_unique_words = udf(count_unique_words, IntegerType())\n",
    "\n",
    "table_df = table_df.withColumn(\"count_unique_words\", udf_count_unique_words(\"body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "words_unique_per_tag = var_per_tag('count_unique_words')\n",
    "words_unique_per_tag_list = list(zip(*words_unique_per_tag))[0]\n",
    "print(words_unique_per_tag)\n",
    "plot_var_per_tag(words_unique_per_tag)\n",
    "words_unique_avg_per_tag = var_avg_per_tag('count_unique_words', words_unique_per_tag_list[:100])\n",
    "print(words_unique_avg_per_tag)\n",
    "plot_var_per_tag(words_unique_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import string\n",
    "import nltk\n",
    "nltk.download('stopwords')\n",
    "from nltk.corpus import stopwords\n",
    "from nltk.stem import PorterStemmer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stop_words = stopwords.words('english')\n",
    "st = PorterStemmer()\n",
    "\n",
    "def count_words2(x):\n",
    "    x = x.lower().translate(str.maketrans('', '', string.punctuation))\n",
    "    return len([st.stem(i) for i in x.split() if i not in stop_words])\n",
    "udf_count_words2 = udf(count_words2, IntegerType())\n",
    "\n",
    "table_df = table_df.withColumn(\"count_words2\", udf_count_words2(\"body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "words2_per_tag = var_per_tag('count_words2')\n",
    "words2_per_tag_list = list(zip(*words2_per_tag))[0]\n",
    "print(words2_per_tag)\n",
    "plot_var_per_tag(words2_per_tag)\n",
    "words2_avg_per_tag = var_avg_per_tag('count_words2', words2_per_tag_list[:100])\n",
    "print(words2_avg_per_tag)\n",
    "plot_var_per_tag(words2_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# count unique words\n",
    "def count_unique_words2(x):\n",
    "    return len(set(str(x).lower().split(\" \")))\n",
    "udf_count_unique_words2 = udf(count_unique_words2, IntegerType())\n",
    "\n",
    "table_df = table_df.withColumn(\"count_unique_words2\", udf_count_unique_words2(\"body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "words2_unique_per_tag = var_per_tag('count_unique_words2')\n",
    "words2_unique_per_tag_list = list(zip(*words2_unique_per_tag))[0]\n",
    "print(words2_unique_per_tag)\n",
    "plot_var_per_tag(words2_unique_per_tag)\n",
    "words2_unique_avg_per_tag = var_avg_per_tag('count_unique_words2', words2_unique_per_tag_list[:100])\n",
    "print(words2_unique_avg_per_tag)\n",
    "plot_var_per_tag(words2_unique_avg_per_tag)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "stop_words = stopwords.words('english')\n",
    "st = PorterStemmer()\n",
    "regex = re.compile('[^a-zA-Z ]')\n",
    "\n",
    "def clean_text(x):\n",
    "    return ' '.join([st.stem(i) for i in regex.sub('', x).lower().translate(str.maketrans('', '', string.punctuation)).split() if i not in stop_words])\n",
    "udf_clean_text = udf(clean_text, StringType())\n",
    "\n",
    "table_df = table_df.withColumn(\"body2\", udf_clean_text(\"body\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "posts = table_df.select('body2').collect()\n",
    "posts = [i[0] for i in posts]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.decomposition import LatentDirichletAllocation\n",
    "from sklearn.feature_extraction.text import CountVectorizer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "NUM_TOPICS = 10\n",
    " \n",
    "vectorizer = CountVectorizer(min_df=5, max_df=0.9)\n",
    "posts_vectorized = vectorizer.fit_transform(posts)\n",
    "\n",
    "lda_model = LatentDirichletAllocation(n_components=NUM_TOPICS, max_iter=10, learning_method='online')\n",
    "lda_model.fit_transform(posts_vectorized)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_names = vectorizer.get_feature_names()\n",
    "for topic_idx, topic in enumerate(lda_model.components_):\n",
    "    topic = topic/topic.sum()\n",
    "    message = \"Topic #%d: \\n\" % topic_idx\n",
    "    message += \" + \".join([\"{:f}\".format(topic[i]) + '*' + feature_names[i] for i in topic.argsort()[:-20 - 1:-1]])\n",
    "    print(message)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyLDAvis.sklearn\n",
    " \n",
    "pyLDAvis.enable_notebook()\n",
    "panel = pyLDAvis.sklearn.prepare(lda_model, posts_vectorized, vectorizer, mds='tsne')\n",
    "panel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}