{"cells":[{"metadata":{},"cell_type":"markdown","source":"This is a CPU only kernel simply to test dask dataframe's capability to handle dataframes larger than cpu memory.\n\nTakeaways:\n* dask can handle dataframes larger than memory by breaking it down into chunks.\n* dask arrary operations are utilizing multi-threads out of the box.\n* element wise operation such as masking, reduction can be done in reasonable time.\n* groupby-aggregation might be doable but it is too slow to be useful."},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"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 in \n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\nimport dask.dataframe as dd\nimport dask\nimport time\n# Input data files are available in the \"../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# Any results you write to the current directory are saved as output.","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"!ls -lsh ../input/PLAsTiCC-2018/test_set.csv","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# monitor cpu memory usage\n!free -g","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n# kernel died running the following pandas command due to OOM\n#df = pd.read_csv('../input/PLAsTiCC-2018/test_set.csv') \n\ndf = dd.read_csv('../input/PLAsTiCC-2018/test_set.csv') ","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"dask is lazy so the dataframe is NOT read yet but we can still access the header instantly"},{"metadata":{"trusted":true},"cell_type":"code","source":"df","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"As shown above, dask breaks down the big dataframe into 310 chunks."},{"metadata":{"trusted":true},"cell_type":"code","source":"df.head() # dask just reads the head","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ndf.shape # dask is not getting the actual shape since it is lazy","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"The number of rows is a `delayed` object and the number of columns is trivial to get, which is `6`"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ndask.compute(df.shape)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"It should be noted that the wall time is roughly 25% of the total CPU time, indicating that `dask` is using 4 threads to do things in parallel."},{"metadata":{},"cell_type":"markdown","source":"Let's caculate the mean value of a column."},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n# simple column-wise reduction operations\ndf['flux'].mean().compute() # returns a scalar","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"By using the `compute` method, dask reads the big csv file chunk by chunk and calculate the mean value. "},{"metadata":{"trusted":true},"cell_type":"code","source":"!free -g","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Notice that the CPU memory usage is not increased since dask has already released memory of all the intermediate variables in the process of calculating `mean`"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ndf_sample = df.loc[df.object_id==13].compute()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Next let's do a grouby aggregation"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\n# it runs for more than 9 hours and is killed by kaggle.\n#flux_stats_of_each_mjd = df.groupby('mjd').agg({'flux':['std']}).compute()\n# This will return a pandas dataframe\n\n#flux_stats_of_each_mjd.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#print(type(flux_stats_of_each_mjd),flux_stats_of_each_mjd.shape)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"`flux_stats_of_each_mjd` is a pandas dataframe. Unfortunately, it is a slow operation that runs for more than 9 hours and killed by kaggle but in theory eventually it should get things done."},{"metadata":{},"cell_type":"markdown","source":"You might think that we need to the whole dataframe into memory to do the groupby aggregation. However, dask adopts an [apply-concat-apply](https://blog.dask.org/2019/10/08/df-groupby) paradigm, where aggregation is done first for each chunk, and then the aggregated intermediate results are concatenated to form a new dataframe, and aggregated again. "}],"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":1}