Data_PreProcessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# BitCoin Price Prediction using Sentiment analysis on social media \n",
    "## Team Members:\n",
    "* Harish Puvvada      - hp1047\n",
    "* Rohit Suvarna       - rus209\n",
    "* Manoj Sundar Balaji - msb662\n",
    "* Janeil Atul Patel   - jap813\n",
    "\n",
    "Github link: https://github.com/harishpuvvada/BitCoin-Value-Predictor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark as spark\n",
    "import pandas as pd\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.functions import col,udf,monotonically_increasing_id,unix_timestamp,round,avg\n",
    "import re\n",
    "sc = spark.SparkContext()\n",
    "sql = spark.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading tweets dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Skipping line 845142: unexpected end of data\n"
     ]
    }
   ],
   "source": [
    "Twdf1=pd.read_csv('/Users/harishpuvvada/Desktop/PBDA/IPynbSpark/tweetsfinal1.csv',error_bad_lines=False,engine = 'python',header = None) \n",
    "# Twdf2=pd.read_csv('/Users/harishpuvvada/Desktop/PBDA/IPynbSpark/tweetsfinal2.csv',error_bad_lines=False,engine = 'python',header = None) \n",
    "# Twdf3=pd.read_csv('/Users/harishpuvvada/Desktop/PBDA/IPynbSpark/tweetsfinal3.csv',error_bad_lines=False,engine = 'python',header = None) \n",
    "# Twdf4=pd.read_csv('/Users/harishpuvvada/Desktop/PBDA/IPynbSpark/tweetsfinal4.csv',error_bad_lines=False,engine = 'python',header = None) \n",
    "# Twdf2 = Twdf2.drop([2,3], axis=1)\n",
    "# Twdf3 = Twdf3.drop([2,3], axis=1)  # this is to drop the gps coordinates in the dataset\n",
    "# Twdf4 = Twdf3.drop([2,3], axis=1)\n",
    "# print(Twdf1.head())\n",
    "# print(Twdf3.head())\n",
    "# print(Twdf2.head())\n",
    "# TwDF = pd.concat([Twdf1,Twdf2,Twdf3,Twdf4])\n",
    "TwDF = Twdf1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading Bitcoin prices dataset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "BtcDF=pd.read_csv('/Users/harishpuvvada/Desktop/PBDA/IPynbSpark/BitCoinPrice.csv',error_bad_lines=False,engine = 'python',header = None) \n",
    "FullDataTw=sql.createDataFrame(TwDF)\n",
    "FullDataBtc=sql.createDataFrame(BtcDF) #creating pandas df and then changing it to pyspark df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "FullDataTw = FullDataTw.dropna() #getting rid of full empty rows\n",
    "#print(FullDataTw.count())\n",
    "#print(FullDataBtc.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "FullDataTw.select(monotonically_increasing_id().alias(\"rowId\"),\"*\")\n",
    "FullDataTw = FullDataTw.withColumnRenamed('0', 'DateTime') #setting column names of Twitter dataset\n",
    "FullDataTw = FullDataTw.withColumnRenamed('1', 'Tweet')\n",
    "FullDataBtc = FullDataBtc.withColumnRenamed('0', 'DateTime') #setting column names of Bitcoin price dataset\n",
    "FullDataBtc = FullDataBtc.withColumnRenamed('1', 'Price')\n",
    "FullDataBtc = FullDataBtc.filter(FullDataBtc.DateTime != 'Date') #to get rid of first row with the header"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pre-Processing Twitter dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "Tw_samp = FullDataTw  #.limit(100) #taking sample of 100 rows and working on it otherwise remove the limit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+\n",
      "|            DateTime|               Tweet|       CleanedTweets|\n",
      "+--------------------+--------------------+--------------------+\n",
      "|Thu Nov 09 17:43:...|RT @Forbes: The F...|The Failure of Se...|\n",
      "|Thu Nov 09 17:43:...|RT @mindstatex: L...|Lots of love from...|\n",
      "|Thu Nov 09 17:43:...|RT @FernandoHuama...|Warning Built in ...|\n",
      "+--------------------+--------------------+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import preprocessor as p #cleaning each tweet using tweet-preprocessor like removing hashtags,urls,emojis....\n",
    "def function_udf(input_str):\n",
    "    input_str = re.sub(r'RT', '', input_str)\n",
    "    p.set_options(p.OPT.URL, p.OPT.EMOJI,p.OPT.MENTION)\n",
    "    input_str = p.clean(input_str)\n",
    "    return ' '.join(re.sub(\"(@[A-Za-z0-9]+)|([^0-9A-Za-z \\t])|(\\w+:\\/\\/\\S+)\", \" \", input_str).split())\n",
    "func_udf = udf(function_udf, StringType())\n",
    "CleanDF = Tw_samp.withColumn('CleanedTweets', func_udf(Tw_samp['Tweet']))\n",
    "CleanDF.show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sentiment analysis using Text blob and Vader packages"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<p>\n",
    "<span style=\"color:blue\">\n",
    "Note: install nltk packages before executing the below cell<br>\n",
    "Our analysis: <br>\n",
    "    Vader takes 0.0005 sec for each observation <br>\n",
    "    Text Blob takes 3.33 sec for each observation <br>\n",
    "</span>\n",
    "</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Textblob code\n",
    "# from textblob import TextBlob  #passing cleaned tweets and getting a sentiment score for each tweet\n",
    "# from textblob.sentiments import NaiveBayesAnalyzer\n",
    "# def senti_score_udf(input_str):\n",
    "#     PSanalysis = TextBlob(input_str)\n",
    "#     analysis = TextBlob(input_str,analyzer=NaiveBayesAnalyzer())\n",
    "#     polarity = PSanalysis.sentiment.polarity\n",
    "#     subjectivity = PSanalysis.sentiment.subjectivity\n",
    "#     classification = analysis.sentiment.classification\n",
    "#     p_pos = analysis.sentiment.p_pos\n",
    "#     p_neg = analysis.sentiment.p_neg\n",
    "#     return [polarity,subjectivity,classification,p_pos,p_neg] #subjectivity, polarity, p_neg, classification\n",
    "# func_udf2 = udf(senti_score_udf, ArrayType(FloatType()))\n",
    "# CleanDF = CleanDF.withColumn('polarity', func_udf2(CleanDF['CleanedTweets'])[0])\n",
    "# CleanDF = CleanDF.withColumn('subj', func_udf2(CleanDF['CleanedTweets'])[1])\n",
    "# CleanDF = CleanDF.withColumn('class', func_udf2(CleanDF['CleanedTweets'])[2])\n",
    "# CleanDF = CleanDF.withColumn('p_pos', func_udf2(CleanDF['CleanedTweets'])[3])\n",
    "# CleanDF = CleanDF.withColumn('p_neg', func_udf2(CleanDF['CleanedTweets'])[4])\n",
    "# CleanDF.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+\n",
      "|            DateTime|               Tweet|       CleanedTweets|p_neg|p_neu|p_pos| p_comp|\n",
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+\n",
      "|Thu Nov 09 17:43:...|RT @Forbes: The F...|The Failure of Se...|0.204|0.617|0.179|-0.1027|\n",
      "|Thu Nov 09 17:43:...|RT @mindstatex: L...|Lots of love from...|  0.0| 0.56| 0.44|  0.875|\n",
      "|Thu Nov 09 17:43:...|RT @FernandoHuama...|Warning Built in ...| 0.13| 0.87|  0.0|  -0.34|\n",
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer\n",
    "analyser = SentimentIntensityAnalyzer()\n",
    "def senti_score_udf(sentence):\n",
    "    snt = analyser.polarity_scores(sentence)\n",
    "    return ([snt['neg'], snt['neu'], snt['pos'], snt['compound']])\n",
    "func_udf2 = udf(senti_score_udf, ArrayType(FloatType()))\n",
    "CleanDF = CleanDF.withColumn('p_neg', func_udf2(CleanDF['CleanedTweets'])[0])\n",
    "CleanDF = CleanDF.withColumn('p_neu', func_udf2(CleanDF['CleanedTweets'])[1])\n",
    "CleanDF = CleanDF.withColumn('p_pos', func_udf2(CleanDF['CleanedTweets'])[2])\n",
    "CleanDF = CleanDF.withColumn('p_comp', func_udf2(CleanDF['CleanedTweets'])[3])\n",
    "CleanDF.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+-------------------+-------------------+\n",
      "|            DateTime|               Tweet|       CleanedTweets|p_neg|p_neu|p_pos| p_comp|         DateTime_c|    DateTime_casted|\n",
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+-------------------+-------------------+\n",
      "|Thu Nov 09 17:43:...|RT @Forbes: The F...|The Failure of Se...|0.204|0.617|0.179|-0.1027|2017-11-09 17:43:41|2017-11-09 17:43:41|\n",
      "|Thu Nov 09 17:43:...|RT @mindstatex: L...|Lots of love from...|  0.0| 0.56| 0.44|  0.875|2017-11-09 17:43:40|2017-11-09 17:43:40|\n",
      "|Thu Nov 09 17:43:...|RT @FernandoHuama...|Warning Built in ...| 0.13| 0.87|  0.0|  -0.34|2017-11-09 17:43:39|2017-11-09 17:43:39|\n",
      "+--------------------+--------------------+--------------------+-----+-----+-----+-------+-------------------+-------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def Tw_Time_format(stri):  #manipulating and casting the strings(DateTime) of tweets dataframe to timestamps\n",
    "    dic = {'Nov':'11','Oct':'10'}\n",
    "    ans = ''\n",
    "    ans += stri[-4:]+'-'+ dic[stri[4:7]]+'-'+stri[8:19]\n",
    "    return ans\n",
    "func_udf3 = udf(Tw_Time_format,StringType())\n",
    "CleanDF = CleanDF.withColumn('DateTime_c', func_udf3(CleanDF['DateTime']))\n",
    "CleanDF = CleanDF.withColumn(\"DateTime_casted\",CleanDF['DateTime_c'].cast(TimestampType()))\n",
    "CleanDF.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "|          Date_Time|      Cleaned_Tweets|p_neg|p_neu|p_pos| p_comp|\n",
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "|2017-11-09 17:43:41|The Failure of Se...|0.204|0.617|0.179|-0.1027|\n",
      "|2017-11-09 17:43:40|Lots of love from...|  0.0| 0.56| 0.44|  0.875|\n",
      "|2017-11-09 17:43:39|Warning Built in ...| 0.13| 0.87|  0.0|  -0.34|\n",
      "|2017-11-09 17:43:39|Join our telegram...|  0.0|0.845|0.155|  0.296|\n",
      "|2017-11-09 17:43:39|DIGAF FLOAT 16M T...|  0.0|  1.0|  0.0|    0.0|\n",
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "FinalTw = CleanDF.selectExpr(\"DateTime_casted as Date_Time\", \"CleanedTweets as Cleaned_Tweets\", \"p_neg\",\"p_neu\",\"p_pos\",\"p_comp\")\n",
    "FinalTw.show(5) #selecting necessary columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pre-Processing Bitcoin dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-------+------------------+\n",
      "|     DateTime|  Price|  Cleaned_BTC_Time|\n",
      "+-------------+-------+------------------+\n",
      "|10/30/17 0:00|6123.21|2017-10-30 0:00:00|\n",
      "|10/30/17 1:00|6131.35|2017-10-30 1:00:00|\n",
      "|10/30/17 2:00|6114.17|2017-10-30 2:00:00|\n",
      "|10/30/17 3:00|6153.11|2017-10-30 3:00:00|\n",
      "|10/30/17 4:00|6151.09|2017-10-30 4:00:00|\n",
      "+-------------+-------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from datetime import datetime \n",
    "from dateutil import parser\n",
    "def Btc_Time_format(input_str): #manipulating and casting the strings(DateTime) of BTC dataframe to timestamps\n",
    "    input_str = re.sub(r'/17','', input_str)\n",
    "    input_str = '2017-'+ input_str\n",
    "    input_str = re.sub(r'/', '-', input_str)\n",
    "    input_str += ':00'\n",
    "    return input_str[:10]+\"\"+input_str[10:]\n",
    "func_udf = udf(Btc_Time_format, StringType())\n",
    "FullDataBtc = FullDataBtc.withColumn('Cleaned_BTC_Time', func_udf(FullDataBtc['DateTime']))\n",
    "FullDataBtc.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------+\n",
      "|          Date_Time|  Price|\n",
      "+-------------------+-------+\n",
      "|2017-10-30 00:00:00|6123.21|\n",
      "|2017-10-30 01:00:00|6131.35|\n",
      "|2017-10-30 02:00:00|6114.17|\n",
      "|2017-10-30 03:00:00|6153.11|\n",
      "|2017-10-30 04:00:00|6151.09|\n",
      "+-------------------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "CleandfBtc = FullDataBtc.withColumn(\"Cleaned_BTC_Time_New\",FullDataBtc['Cleaned_BTC_Time'].cast(TimestampType()))\n",
    "FinalBtc = CleandfBtc.selectExpr(\"Cleaned_BTC_Time_New as Date_Time\", \"Price\")\n",
    "FinalBtc = FinalBtc.withColumn(\"Price\",FinalBtc['Price'].cast(DoubleType()))\n",
    "FinalBtc.show(5)#In this cell, casting to timesstamp, changing col names and casting price type to double"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dataframes Look like this..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date_Time: timestamp (nullable = true)\n",
      " |-- Cleaned_Tweets: string (nullable = true)\n",
      " |-- p_neg: float (nullable = true)\n",
      " |-- p_neu: float (nullable = true)\n",
      " |-- p_pos: float (nullable = true)\n",
      " |-- p_comp: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "FinalTw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date_Time: timestamp (nullable = true)\n",
      " |-- Price: double (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "672"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "FinalBtc.printSchema()\n",
    "FinalBtc.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Truncating timestamps to hours and then grouping them by hour"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "|          Date_Time|      Cleaned_Tweets|p_neg|p_neu|p_pos| p_comp|\n",
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "|2017-11-09 23:00:00|The Failure of Se...|0.204|0.617|0.179|-0.1027|\n",
      "|2017-11-09 23:00:00|Lots of love from...|  0.0| 0.56| 0.44|  0.875|\n",
      "|2017-11-09 23:00:00|Warning Built in ...| 0.13| 0.87|  0.0|  -0.34|\n",
      "|2017-11-09 23:00:00|Join our telegram...|  0.0|0.845|0.155|  0.296|\n",
      "|2017-11-09 23:00:00|DIGAF FLOAT 16M T...|  0.0|  1.0|  0.0|    0.0|\n",
      "+-------------------+--------------------+-----+-----+-----+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dt_truncated = ((round(unix_timestamp(col('Date_Time')) / 3600) * 3600).cast('timestamp'))\n",
    "FinalTw = FinalTw.withColumn('dt_truncated', dt_truncated)\n",
    "FinalTw = FinalTw.selectExpr(\"dt_truncated as Date_Time\",\"Cleaned_Tweets\",\"p_neg\",\"p_neu\",\"p_pos\",\"p_comp\")\n",
    "UTC = ((unix_timestamp(col('Date_Time'))+ 5*60*60).cast('timestamp'))\n",
    "FinalTw = FinalTw.withColumn('UTC', UTC)\n",
    "FinalTw = FinalTw.selectExpr(\"UTC as Date_Time\",\"Cleaned_Tweets\",\"p_neg\",\"p_neu\",\"p_pos\",\"p_comp\")\n",
    "FinalTw.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "|           DateTime|               P_Neg|             P_Neu|              P_Pos|             P_Comp|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "|2017-11-09 23:00:00|0.035409999862313274|0.8348199963569641|0.12977000258862972|0.23347700215876102|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "FinalTw.registerTempTable(\"temp\")\n",
    "FinalTw_avg = sql.sql(\"SELECT Date_Time As DateTime,AVG(p_neg) as P_Neg,AVG(p_neu) as P_Neu,AVG(p_pos) as P_Pos,AVG(p_comp) as P_Comp FROM temp GROUP BY Date_Time\")\n",
    "#FinalTw_avg = FinalTw.select(\"Date_Time\",\"polarity\",\"subj\",\"p_pos\",\"p_neg\").groupBy(\"Date_Time\").agg(avg(col(\"polarity\",\"subj\",\"p_pos\",\"p_neg\")))\n",
    "FinalTw_avg.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------------------------------------+\n",
      "|          Date_Time|concat_ws( , collect_list(Cleaned_Tweets))|\n",
      "+-------------------+------------------------------------------+\n",
      "|2017-11-09 23:00:00|                      The Failure of Se...|\n",
      "+-------------------+------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#This cell is just to collect all the corpus per hour(for the future work)\n",
    "from pyspark.sql import functions as f\n",
    "df_with_text = FinalTw.groupby(\"Date_Time\").agg(f.concat_ws(\" \", f.collect_list(FinalTw.Cleaned_Tweets)))\n",
    "df_with_text.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "|           DateTime|               P_Neg|             P_Neu|              P_Pos|             P_Comp|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "|2017-11-09 23:00:00|0.035409999862313274|0.8348199963569641|0.12977000258862972|0.23347700215876102|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "FinalTw_avg.count()\n",
    "from pyspark.sql.functions import *\n",
    "df_sort = FinalTw_avg.sort(asc(\"Date_Time\"))\n",
    "df_sort.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Joining twitter and bitcoin dataframes by DateTime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+------------------+-------------------+-------------------+-------+\n",
      "|           DateTime|               P_Neg|             P_Neu|              P_Pos|             P_Comp|  Price|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+-------+\n",
      "|2017-11-09 23:00:00|0.035409999862313274|0.8348199963569641|0.12977000258862972|0.23347700215876102|7134.47|\n",
      "+-------------------+--------------------+------------------+-------------------+-------------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "FinalTw_avg.registerTempTable(\"avgs\")\n",
    "FinalBtc.registerTempTable(\"prices\")\n",
    "results = sql.sql(\"SELECT DateTime, P_Neg, P_Neu, P_Pos, P_Comp, Price FROM avgs JOIN prices ON avgs.DateTime = prices.Date_Time order by avgs.DateTime\")\n",
    "#results = results.selectExpr(\"DateTime\",\"avg(polarity)\",\"avg(subj)\",\"avg(p_pos)\",\"avg(p_neg)\",\"Price\") Use this line if you are using text blob package\n",
    "results.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "results.count()  # Final size of dataset after joining bitcoin and twitter dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "results.repartition(1).write.csv(\"DataforModelExec.csv\") #this will write df to single csv instead of writing diff csv acc to partitions "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Now refer to LSTM notebook for the timeseries analysis"
   ]
  }
 ],
 "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}