spark_version.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook shows how to transform tweets in line-oriented JSON files to data that's useful for topic modeling!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:10:45.848196Z",
     "start_time": "2020-05-17T18:10:35.467274Z"
    }
   },
   "outputs": [],
   "source": [
    "# this is big data\n",
    "# use Spark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import element_at, col, coalesce, length, max as max_, concat_ws\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:19:02.641636Z",
     "start_time": "2020-05-17T18:10:45.872941Z"
    }
   },
   "outputs": [],
   "source": [
    "# the tweet files, about 6gb each, are compressed into .gz files, which spark can read\n",
    "df = spark.read.json(\"compressed_tweets/*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:19:03.330885Z",
     "start_time": "2020-05-17T18:19:03.077291Z"
    },
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "# concatenate the mentions, hashtags, and urls into pandas-transformation friendly single strings\n",
    "# pyarrow, the Pandas conversion engine, can't handle arrays from Spark yet\n",
    "df = df.withColumn(\"mentions\", concat_ws(\" \", col(\"entities.user_mentions.screen_name\")))\n",
    "df = df.withColumn(\"hashtags\", concat_ws(\" \", \"entities.hashtags.text\"))\n",
    "df = df.withColumn(\"urls\", concat_ws(\" \", \"entities.urls.display_url\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:19:03.553064Z",
     "start_time": "2020-05-17T18:19:03.333034Z"
    }
   },
   "outputs": [],
   "source": [
    "# limit to languages that I speak\n",
    "df = df.where(df.lang.isin({\"en\", \"tr\", \"fr\"}))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:19:03.611954Z",
     "start_time": "2020-05-17T18:19:03.557042Z"
    }
   },
   "outputs": [],
   "source": [
    "# select only a few useful columns\n",
    "# a full tweet object contains a BUNCH of stuff that you can read about here:\n",
    "# https://developer.twitter.com/en/docs/tweets/data-dictionary/overview/intro-to-tweet-json\n",
    "df = df.select(          \n",
    "              \"created_at\",\n",
    "              \"full_text\",\n",
    "              \"lang\",\n",
    "              \"favorite_count\", \n",
    "              \"retweet_count\", \n",
    "              \"place.country\", \n",
    "              \"place.place_type\", \n",
    "              \"place.full_name\",\n",
    "              \"mentions\",\n",
    "              \"urls\",\n",
    "              \"hashtags\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:42:06.733586Z",
     "start_time": "2020-05-17T18:37:03.132141Z"
    }
   },
   "outputs": [],
   "source": [
    "# write json\n",
    "# I had best results with pandas-spark intercompatibility using JSONs\n",
    "df.write.json(\"parsed_tweets3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-05-17T18:45:27.250970Z",
     "start_time": "2020-05-17T18:45:10.411332Z"
    }
   },
   "outputs": [],
   "source": [
    "# test reading with pandas\n",
    "# spark writes line-oriented JSONs - one complete JSON object per line\n",
    "import pandas as pd\n",
    "test_df = pd.read_json(\"parsed_tweets3/part-00000-1d9e27cf-c47e-43c5-b56c-c3c8448494c4-c000.json\", lines = True)"
   ]
  }
 ],
 "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}