Datamodeling.ipynb

{"cells":[{"cell_type":"code","source":["df = spark.createDataFrame([\n        (1, 144.5, 5.9, 33, 'M'),\n        (2, 167.2, 5.4, 45, 'M'),\n        (3, 124.1, 5.2, 23, 'F'),\n        (4, 144.5, 5.9, 33, 'M'),\n        (5, 133.2, 5.7, 54, 'F'),\n        (3, 124.1, 5.2, 23, 'F'),\n        (5, 129.2, 5.3, 42, 'M'),\n    ], ['id', 'weight', 'height', 'age', 'gender'])"],"metadata":{},"outputs":[],"execution_count":1},{"cell_type":"code","source":["display(df)"],"metadata":{},"outputs":[],"execution_count":2},{"cell_type":"code","source":["df.select('age').distinct().collect()"],"metadata":{},"outputs":[],"execution_count":3},{"cell_type":"code","source":["df.distinct().count()"],"metadata":{},"outputs":[],"execution_count":4},{"cell_type":"code","source":["#this will remove if entire row is same including id\ndf = df.dropDuplicates()"],"metadata":{},"outputs":[],"execution_count":5},{"cell_type":"code","source":["display(df)"],"metadata":{},"outputs":[],"execution_count":6},{"cell_type":"code","source":["df.select(['weight','height','age','gender']).distinct().collect()"],"metadata":{},"outputs":[],"execution_count":7},{"cell_type":"code","source":["new_df = df.select([c for c in df.columns if c!= 'id']).distinct()"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"code","source":[""],"metadata":{},"outputs":[],"execution_count":9},{"cell_type":"code","source":["import pyspark.sql.functions as fn\n\ndisplay(df)"],"metadata":{},"outputs":[],"execution_count":10},{"cell_type":"code","source":["df.agg( fn.count('id').alias('count'), fn.countDistinct('gender').alias('distinct')).show()"],"metadata":{},"outputs":[],"execution_count":11},{"cell_type":"code","source":["df.withColumn('new_id', fn.monotonically_increasing_id()).show()"],"metadata":{},"outputs":[],"execution_count":12},{"cell_type":"code","source":["df_miss = spark.createDataFrame([\n        (1, 143.5, 5.6, 28,   'M',  100000),\n        (2, 167.2, 5.4, 45,   'M',  None),\n        (3, None , 5.2, None, None, None),\n        (4, 144.5, 5.9, 33,   'M',  None),\n        (5, 133.2, 5.7, 54,   'F',  None),\n        (6, 124.1, 5.2, None, 'F',  None),\n        (7, 129.2, 5.3, 42,   'M',  76000),\n    ], ['id', 'weight', 'height', 'age', 'gender', 'income'])"],"metadata":{},"outputs":[],"execution_count":13},{"cell_type":"code","source":["df_miss.rdd.map(lambda row: (row['id'], sum([ c == None for c in row ]))).collect()"],"metadata":{},"outputs":[],"execution_count":14},{"cell_type":"code","source":["df_miss.where('id == 3').show()"],"metadata":{},"outputs":[],"execution_count":15},{"cell_type":"code","source":["#Now, find out what is the missing data percentage\n# The * argument to the count() method instruct the method to count all row\n# fn.count('age') - returns count of non-None elements\n# * before the list - this that .agg() method to treat the list as a set of different paramerts\ndf_miss.agg( * [ ( 1 - (fn.count(c)/ fn.count('*'))).alias(c+'_missing')   for c in df_miss.columns] ).show()"],"metadata":{},"outputs":[],"execution_count":16},{"cell_type":"code","source":["def f(name,age):\n  print name,age\n\nf(*['abc',33])"],"metadata":{},"outputs":[],"execution_count":17},{"cell_type":"code","source":["# remove rows which have more than 3 entries null\ndf_miss.dropna(thresh=3).show()"],"metadata":{},"outputs":[],"execution_count":18},{"cell_type":"code","source":["df_miss.select([c for c in df_miss.columns if c != 'income']).show()"],"metadata":{},"outputs":[],"execution_count":19},{"cell_type":"code","source":["# Find outliers - observations which deviate from the distribution of rest of the sample\ndf_outliers = spark.createDataFrame([\n        (1, 143.5, 5.3, 28),\n        (2, 154.2, 5.5, 45),\n        (3, 342.3, 5.1, 99),\n        (4, 144.5, 5.5, 33),\n        (5, 133.2, 5.4, 54),\n        (6, 124.1, 5.1, 21),\n        (7, 129.2, 5.3, 42),\n    ], ['id', 'weight', 'height', 'age'])"],"metadata":{},"outputs":[],"execution_count":20},{"cell_type":"code","source":["cols = ['weight', 'height', 'age']\nbounds = {}\n\nfor col in cols:\n  qualtiles = df_outliers.approxQuantile(col, [0.25, 0.75], 0.05)\n  \n  IQR = qualtiles[1] - qualtiles[0]\n  \n  bounds[col] = [\n    qualtiles[0] - IQR * 1.5,\n    qualtiles[1] - IQR * 1.5\n  ]"],"metadata":{},"outputs":[],"execution_count":21},{"cell_type":"code","source":["bounds"],"metadata":{},"outputs":[],"execution_count":22},{"cell_type":"code","source":["qualtiles"],"metadata":{},"outputs":[],"execution_count":23},{"cell_type":"code","source":[""],"metadata":{},"outputs":[],"execution_count":24}],"metadata":{"name":"Datamodeling","notebookId":2299846101781698},"nbformat":4,"nbformat_minor":0}