{"cells":[{"metadata":{"trusted":true,"_kg_hide-input":true,"_kg_hide-output":true},"cell_type":"code","source":"!pip install pyspark","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"\n# What for?\n\nIn this Notebook, I gone through how to use Pyspark to load the data and creating Data gen without out running out of memory it will help in the case of training large models within the the kaggle resource itself"},{"metadata":{},"cell_type":"markdown","source":"# Why Pyspark?\n\nIn Apache Spark, a DataFrame is a distributed collection of rows under named columns. In simple terms, it is same as a table in relational database or an Excel sheet with Column headers. It also shares some common characteristics with RDD:\n\n* **Immutable in nature** : We can create DataFrame / RDD once but can’t change it. And we can transform a DataFrame / RDD after applying transformations.\n* **Lazy Evaluations**: Which means that a task is not executed until an action is performed.\n* **Distributed**: RDD and DataFrame both are distributed in nature.\n\nMy first exposure to DataFrames was when I learnt about Pandas. Today, it is difficult for me to run my data science workflow with out Pandas DataFrames. So, when I saw similar functionality in Apache Spark, I was excited about the possibilities it opens up!\n\n click hear to learn more..."},{"metadata":{},"cell_type":"markdown","source":"Here the main purpose of pyspark is all because of it's lazzy evalution property.\n\n**Pandas** load the whole data into memory when we do some transformation to our data which was already loaded into our memory, will hurts the memory further.\n\n**RDD's** are useful because they allow users to process data at the \"row\" level without having to load all data into memory.\n\n### Saves Computation and increases Speed\nSpark Lazy Evaluation plays a key role in saving calculation overhead. Since only necessary values get compute.\n\n### Reduces Complexities\nThe two main complexities of any operation are time and space complexity. Using pySpark lazy evaluation we can overcome both. Since we do not execute every operation. The action is triggered only when the data is required, it reduces overhead."},{"metadata":{},"cell_type":"markdown","source":"In Our case, Data gen outputs the chunk of data that we need for training. so, we no need to load the whole stuff in the memory we can load it on demand basis."},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import pyspark\nimport gc\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import isnan, when, count, col, lit, udf\nfrom pyspark.sql.types import FloatType, StructType, StructField, IntegerType, Row\nimport pandas as pd","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"data = pd.read_csv(\"../input/riiid-test-answer-prediction/train.csv\", usecols=['user_id'])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"user_ids = data.user_id.value_counts().reset_index().values","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"user_ids.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"del data\ngc.collect()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"\nspark = SparkSession.builder.appName(\"SparkByExamples.com\").getOrCreate()\n\ndf = spark.read.csv(\"../input/riiid-test-answer-prediction/train.csv\",inferSchema = True, header = True)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df.printSchema()","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"df.columns","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df = df.repartition('user_id')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df.rdd.getNumPartitions()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df2 = df.withColumn('prior_question_elapsed_time', when(col('prior_question_elapsed_time').isNull(), when(col('content_type_id') == 0, lit(0)).otherwise(lit(-1))).otherwise(df.prior_question_elapsed_time))\ndf2 = df2.withColumn(\"prior_question_had_explanation\", df2[\"prior_question_had_explanation\"].cast(FloatType()))\ndf2 = df2.withColumn('prior_question_had_explanation', when(col('prior_question_had_explanation').isNull(), lit(-1.0)).otherwise(df2.prior_question_had_explanation))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"questions_data = pd.read_csv('../input/riiid-test-answer-prediction/questions.csv')\nlecture_data = pd.read_csv('../input/riiid-test-answer-prediction/lectures.csv')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"qution_dict = {}\n\nfor i in range(len(questions_data)):\n qution_dict[questions_data.question_id[i]] = [questions_data.tags[i], questions_data.part[i], questions_data.bundle_id[i]]\n\nlecture_dict = {}\n\nfor i in range(len(lecture_data)):\n lecture_dict[lecture_data.lecture_id[i]] = [lecture_data.tag[i], lecture_data.part[i], -1]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"del questions_data, lecture_data","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def leg_que_merge(content_id, content_type_id):\n if content_type_id == 0:\n to_return = qution_dict[content_id]\n try:\n tag = [float(i) for i in to_return[0].split(\" \")]\n except:\n tag = [-1.0]\n dif = 6 - len(tag)\n tag.extend([-1.0]*dif)\n part = float(to_return[1])\n bundle = float(to_return[2])\n return Row('part', 'tag1', 'tag2', 'tag3', 'tag4', 'tag5', 'tag6', 'bundle')(part, tag[0], tag[1], tag[2], tag[3], tag[4], tag[5], bundle)\n else:\n to_return = lecture_dict[content_id]\n try:\n tag = [float(i) for i in to_return[0].split(\" \")]\n except:\n tag = [-1.0]\n dif = 6 - len(tag)\n tag.extend([-1.0]*dif)\n part = float(to_return[1])\n bundle = float(to_return[2])\n return Row('part', 'tag1', 'tag2', 'tag3', 'tag4', 'tag5', 'tag6', 'bundle')(part, tag[0], tag[1], tag[2], tag[3], tag[4], tag[5], bundle)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"schema = StructType([\n StructField(\"part\", FloatType(), False),\n StructField(\"tag1\", FloatType(), False),\n StructField(\"tag2\", FloatType(), False),\n StructField(\"tag3\", FloatType(), False),\n StructField(\"tag4\", FloatType(), False),\n StructField(\"tag5\", FloatType(), False),\n StructField(\"tag6\", FloatType(), False),\n StructField(\"bundle\", FloatType(), False)])\n\nmapper = udf(leg_que_merge, schema)\n\ndf2 = df2.withColumn(\"Output\", mapper(df2['content_id'], df2['content_type_id']))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"e_features = df2.select(\"user_id\", \"content_type_id\", \"Output.*\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"e_features.show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df3 = df.select(\"user_id\",\"content_id\", \"content_type_id\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df3 = df3.repartition(60)\ndf3.explain()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def data_generatore(include_lecutes = False):\n for i in user_ids:\n dat = df3.filter(col('user_id') == float(i))\n yield dat.collect()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%time\nnext(iter(data_generatore()))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Tensorflow"},{"metadata":{},"cell_type":"markdown","source":"# Why tensorflow Dataset API?\n\nTensorflow Dataset API has an ability to load data from plenty of sources like pandas dataframe, Numpy array, csv file, tfrecord file etc., After that it also providing bunch of operations to transform data, to make its consuption optimal and more robust.\n
\n \n
\n \n Achieving peak performance requires an efficient input pipeline that delivers data for the next step before the current step has finished. The tf.data API helps to build flexible and efficient input pipelines\n \nTensorFlow Profiler aims to help users diagnose and fix input pipeline performance issues by finding the performance bottleneck\n\n\n**NOTE:**\nBy design, TensorFlow is based on lazy execution (though we can force eager execution). That means, it does not actually process the data available till it has to. It just gathers all the information that we feed into it. It processes only when we finally ask it to process.\n\nTo learn more..."},{"metadata":{},"cell_type":"markdown","source":"* Loading the csv data using CsvDataset\n\nIn my case I am loading only 5 columns (timestamp, user_id, content_id, content_type_id, answered_correctly). If you want whole column just remove \"select_cols\"\n\nNote:\n Loading Whole Column won't affect the execution speed"},{"metadata":{"trusted":true},"cell_type":"code","source":"import numpy as np\nimport tensorflow as tf\nimport pandas as pd\n\n\nfilenames = '../input/riiid-test-answer-prediction/train.csv'\ndtype = (np.int(), np.float(), np.int(), np.int(), np.float(), np.float(), np.float(), np.float(), np.str(), np.str())\n\ndf_ = tf.data.experimental.CsvDataset(\n filenames, record_defaults = dtype,\n header=True\n)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"next(iter(df_))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_kg_hide-input":true},"cell_type":"code","source":"data = pd.read_csv('../input/riiid-test-answer-prediction/train.csv', usecols = ['user_id'])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"user_ids = data.user_id.value_counts().reset_index().values","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def preprocessing(*args):\n elapsetime = -1.0\n bollean = -1.0\n if args[-2] == b\"\" and args[4] == 0:\n elapsetime = 0.0\n elif args[-2] == b'' and args[4]== 1:\n pass\n else:\n elapsetime = float(args[-2])\n \n if args[-2] == b'':\n pass\n elif args[-2] == 'True':\n bollea = 1.0\n else:\n bollean = 0.0\n return (*args[:8], elapsetime, bollean)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"newdf_ = df_.map(preprocessing, num_parallel_calls = tf.data.experimental.AUTOTUNE)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"next(iter(newdf_))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"\n**Created a Lookup table for qeustions and leture data which is used to map the content id to it's corresponding tags and part**\n\n#### Why we are using Lookup table?\nIn pyspark we had used dictionary as an look up table but in tensorflow we can't use it because of eager execution. As an alternative to dictionary, I had used Look table from tensorflow. In this case,the whole operation that going to be happen in dataset will be in graph format.\n\n\n**note**: If you want to use dictionary, eager model has to disabled at the begining."},{"metadata":{"trusted":true},"cell_type":"code","source":"questions_data = pd.read_csv('../input/riiid-test-answer-prediction/questions.csv')\nlecture_data = pd.read_csv('../input/riiid-test-answer-prediction/lectures.csv')\n\nques_key = []\nques_val = []\nques_idx = []\nfor i in range(len(questions_data)):\n da = questions_data.tags[i]\n if str(da) == 'nan':\n da = [-1.0]\n else:\n da = [float(i) for i in da.split(\" \")]\n diff = 6 - len(da)\n ques_key.append(questions_data.question_id[i])\n ques_val.append([*da, *[-1.0]*diff, questions_data.part[i], questions_data.bundle_id[i]])\n ques_idx.append(i)\n \nlec_key = []\nlec_val = []\nlec_idx = []\nfor i in range(len(lecture_data)):\n da = str(lecture_data.tag[i])\n if da == 'nan':\n da = [-1.0]\n else:\n da = [float(i) for i in da.split(\" \")]\n diff = 6 - len(da)\n lec_key.append(lecture_data.lecture_id[i])\n lec_val.append([*da, *[-1.0]*diff, lecture_data.part[i], -1])\n lec_idx.append(i)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"del lecture_data, questions_data","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"input_tensor = tf.constant([5692])\n\nquestion_values = tf.constant(ques_val)\nlecture_values = tf.constant(lec_val)\n\nque_init = tf.lookup.KeyValueTensorInitializer(tf.constant(ques_key), tf.constant(ques_idx))\nquestion_lookup_table = tf.lookup.StaticHashTable(\n que_init,\n default_value=-1)\n\nlec_init = tf.lookup.KeyValueTensorInitializer(tf.constant(lec_key), tf.constant(lec_idx))\nlecture_lookup_table = tf.lookup.StaticHashTable(\n lec_init,\n default_value=-1)\n\n\nquestion_values[question_lookup_table.lookup(input_tensor).numpy()[0]]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def leg_que_merge(*args):\n content_id = args[3]\n if args[4] == 0:\n va = question_lookup_table.lookup(tf.cast(content_id, dtype = tf.int32))\n to_return = question_values[va]\n return (args[1], args[2],args[7], to_return)\n else:\n va = lecture_lookup_table.lookup(tf.cast(content_id, dtype = tf.int32))\n to_return = lecture_values[va]\n return (args[1], args[2],args[7], to_return)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"newdf_ = newdf_.map(leg_que_merge, num_parallel_calls = tf.data.experimental.AUTOTUNE)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"iterator = iter(newdf_.batch(1))\ntimestamp, user_id, answered_correctly, features = next(iterator)\n\nprint(\"User : \", user_id.numpy()[0], \" \" * 50 + \"TimeStamp : \", timestamp.numpy()[0])\nprint()\nprint(\"Features : \", features.numpy().astype(np.int))\nprint(\"Label : \", answered_correctly.numpy()[0])\n\nprint(\"=\"*80)\ntimestamp, user_id, answered_correctly, features = next(iterator)\n\nprint(\"User : \", user_id.numpy()[0], \" \" * 50 + \"TimeStamp : \", timestamp.numpy()[0])\nprint()\nprint(\"Features : \", features.numpy().astype(np.int))\nprint(\"Label : \", answered_correctly.numpy()[0])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Conclution:\n\nBoth pyspark and tensorfow are working as lazzy execution. The whole data is not pulled into the memory instead we can fetch it on demand. It helps us to avoid out of memory issue when working with pandas.\n\nBut we have a sight over execution speed tensorflow seems to be much faster than pyspark."}],"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":4,"nbformat_minor":4}