{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!pip install pyspark","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import pyspark","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Importing SparkSession\nfrom pyspark.sql import SparkSession","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Starting a pyspark session #\n\nspark = SparkSession.builder.appName('test').getOrCreate()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"spark\n# Only 1 cluster is created\n# When we work in the cloud, pyspark creates multiple clusters for parallel processing the data, which is called as the distributed computing.\n# Spark session is started","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Reading the data from train.csv (file size: 5.45 GB)\nimport time\nt1 = time.time()\ndf = spark.read.csv('../input/riiid-test-answer-prediction/train.csv')\nt2 =time.time()\ndiff = t2-t1\nprint(diff)\n\n# Able to read the 5.45 GB data in just 7 seconds\n# Size of the data : 100 Mn Records","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df.count()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#Pandas is very efficient with small data (usually from 100MB up to 1GB) \n# When reading this huge dataset through pandas, it throws an out of memory error\nimport time\nimport pandas as pd\nt1 = time.time()\ndf1 = pd.read_csv('../input/riiid-test-answer-prediction/train.csv',chunksize=1000000)\nt2 = time.time()\ndiff = t2-t1\nprint(diff)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df.head()\n# It gives the output in the form of column names : c0,c1, c2, c3, c4..","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# While reading the dataset, we can provide header as true, to bring the column names instead of c1,c2, c3...\ndf1 = spark.read.option('header','true').csv('../input/riiid-test-answer-prediction/train.csv')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Just like info provides datatype of all column in pandas \n# In case of pyspark we use printSchema to get datatype of the columns\ndf.printSchema()\n# By default it is taking the string as a datatype for all columns ","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# While reading the dataset, we can provide the schema for all variables, so it takes the appropriate datatypes for all columns and putting header=True to get column names\ndf1 = spark.read.csv('../input/riiid-test-answer-prediction/train.csv',header=True,inferSchema=True)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# It provides the actual data type of all columns #\ndf1.printSchema()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# dtypes als helps to check the datatypes of all columns\ndf1.dtypes","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# top 2 records\ndf1.head(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Column names #\ndf1.columns","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Get the top 6 records of a specific column\ndf1.select('user_id').show(6)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Picking multiple columns at a time\n\ndf1.select(['user_id','user_answer']).show(3)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Summary of all the variables - describe \ndf1.describe().show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Check the distinct categories in a column #\n\nfrom pyspark.sql.functions import countDistinct\nc1 = df1.select(countDistinct('user_answer'))\nc1.show()\n# 5 unique categories","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Making the value counts for each of the categories in the column : user_answer\nc2=df1.groupby('user_answer').count()\nc2.show()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Creating a new column based on the user answers, if a person has provided answer has 2, lets call it as True, elee as False\ndf1=df1.withColumn('user_answer_flag',df1['user_answer']>1)\ndf1.select(['user_answer','user_answer_flag']).show(5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Dropping the columns # ( Drop column : user_answer_flag)\n\ndf1 = df1.drop('user_answer_flag')\ndf1.columns","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Renaming the columns #\ndf1 = df1.withColumnRenamed('user_id','User Id')\ndf1.columns","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# To get the shape of the dataframe #\n\nprint(df1.count(),len(df1.columns))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Check the  missing values in dataframe\nfrom pyspark.sql.functions import isnan, when, count, col\ndf1.select([count(when(col(c).isNull(), c)).alias(c) for c in df1.columns]).show()\n# There are missing values : prior_question_elapsed_time and prior_question_had_explanation","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Parameters in dropping the null records #\n# how ='any' (default), 'all'\n# subset : we can drop the records if there are null values in a specific column\n# thresh : for eg thres 2, it will keep those records which have atleast 2 non null values","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Dropping the rows of the dataframe if there is any missing value in any column of the dataframe\ndf1.na.drop().count()\n# By default , how='any', if it founds any null values in any column, it will drop those rows","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df1.na.drop(how='all').count()\n# If how='all', if it founds all the columns as null for a record, then only it will drop that record","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df1.count()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Handling missing values #\n# Wherever there is a missing value, it will fill missing values with 'Missing'\n\ndf.na.fill('Missing').show(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"df1.show(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filling missing values with mean value #\n\n# We have an imputer function which will calcutate the mean of the entire column, which can be used to impute the missing values\n\nfrom pyspark.ml.feature import Imputer\nimputer = Imputer(inputCols=['prior_question_elapsed_time'],outputCols=[\"{}_imputed\".format(c) for c in ['prior_question_elapsed_time']]).setStrategy('mean')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"imputer.fit(df1).transform(df1).show(2)\n\n# We have created a new column : prior_question_elapsed_time_imputed which has imputed the missing values with the mean values.\n# Futher we can drop the prior_question_elapsed_time column which has missing values","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filtering Operations on the dataframe to fetch the records of dataframe #\n# Filter the records with user_answer>1 \ndf1.filter('user_answer>1').show(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Different way of filtering the same data\ndf1.filter(df1['user_answer']>1).show(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# After filtering showing only 2 columns \ndf1.filter('user_answer>1').select(['user_answer','answered_correctly']).show(2)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Filtering operations using 'and' #\n# User Answer : 2\ndf1.filter((df1['user_answer']>1) & (df1['user_answer']<3)).show(4)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Using the Or (|) operators\n\ndf1.filter((df1['user_answer']==1) | (df1['user_answer']==3)).show(4)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Not Operator (~) :Not having 1\ndf1.filter(~(df1['user_answer']==1)).show(4)","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}