{"cells":[{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"%%time\n\npip install pyspark -q","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"from pyspark.sql import SparkSession, DataFrame, Window, functions as F, types as T\nfrom pyspark.ml.feature import Bucketizer\nimport pandas as pd","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"spark = SparkSession.builder.appName('_').getOrCreate()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def train_to_tmp():\n\n    train_schema = T.StructType([\n        T.StructField('row_id',                          T.LongType(),    False),\n        T.StructField('timestamp',                       T.LongType(),    False),\n        T.StructField('user_id',                         T.IntegerType(), False),\n        T.StructField('content_id',                      T.ShortType(),   False),\n        T.StructField('content_type_id',                 T.ByteType(),    False),\n        T.StructField('task_container_id',               T.ShortType(),   False),\n        T.StructField('user_answer',                     T.ByteType(),    False),\n        T.StructField('answered_correctly',              T.ByteType(),    False),\n        T.StructField('prior_question_elapsed_time',     T.FloatType(),   False),\n        T.StructField('prior_question_had_explanation',  T.BooleanType(), False),\n    ])\n\n    train = spark\\\n    .read.csv('../input/riiid-test-answer-prediction/train.csv',\n              schema = train_schema,\n              enforceSchema = True,\n              header = True)\\\n    .withColumn('prior_question_had_explanation', F.col('prior_question_had_explanation').astype(T.ByteType()))\n    #.na.fill(-1)\n\n    train.write.parquet('tmp', mode = 'overwrite')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ntrain_to_tmp()\ntmp = spark.read.parquet('./tmp')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ntmp\\\n.select(['user_id','timestamp','task_container_id']).distinct()\\\n.withColumn('new_order', \n            F.row_number().over(\n                Window.partitionBy('user_id').orderBy('timestamp')\n            )-1)\\\n.write.parquet('utc', mode = 'overwrite')\nutc = spark.read.parquet('./utc')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ntmp\\\n.join(utc,on=['user_id','timestamp','task_container_id'], how = 'inner')\\\n.write.parquet('tmp_new', mode = 'overwrite')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"!rm -rf  ./tmp\n!rm -rf  ./utc","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ndf = spark.read.parquet('./tmp_new')\n\nuser_map = df\\\n.select('user_id')\\\n.distinct()\\\n.withColumn('r', F.floor(F.rand(0)*10))\\\n.write.parquet('user_map', mode = 'overwrite')\n\nuser_map = spark.read.parquet('./user_map')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nfor i in range(10):\n\n    imap = user_map\\\n    .filter(F.col('r')==i)\\\n    .select('user_id')\n\n    idf = df.join(imap, on = 'user_id', how = 'inner')\n\n    idf.repartition(10).write.parquet('train_'+str(i), mode = 'overwrite')\n    \n    _ = pd.read_parquet('./train_'+str(i))\n    !rm -rf  ./train_*\n    _.to_parquet('df_'+str(i))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"!rm -rf  ./tmp_new\n!rm -rf  ./user_map","execution_count":null,"outputs":[]}],"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}