{"cells":[{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true,"_kg_hide-input":true},"cell_type":"code","source":"import os\nimport sys\n\nimport numpy as np\nimport matplotlib\nimport matplotlib.pyplot as plt\nimport pandas as pd\n\nfrom dask.distributed import Client\nimport dask.dataframe as dd\n\nmatplotlib.style.use(\"dark_background\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_kg_hide-input":true},"cell_type":"code","source":"def plot_hist(df, feature_name, kind='hist', bins=100):\n    df[feature_name].plot(kind='hist', \n                          bins=bins, \n                          figsize=(15, 5), \n                          title=f'Distribution of {feature_name}')\n    plt.show()\n\n\ndef plot_ts(series, figsize=(20, 6), title=None, xlabel=\"\", ylabel=\"\"):\n    \"\"\"\n    Plot Time Series data. The series object should have date or time as index.\n    \n    series: Series object to be plotted.\n    \"\"\"\n    series.plot(figsize=figsize, title=title)\n    plt.xlabel(xlabel)\n    plt.ylabel(ylabel)\n    plt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Goal\n\nIn this notebook, I am going to show **how Dask can be used to explore the train and test files related to various data segments.**\n\nAs per the documentation, each of these file contains ten minutes of logs from ten different sensors arrayed around a volcano.\n\nThere are 4432 data files under the train directory and 4521 files under test directory. As we will observe sooner, each of these files consists of 60K lines. On the disk, size of the files under train and test directory is 30G.\n\nThere are mostly two problems we encounter as a Data Scientist when dealing with such a large volume of data:\n\n1. **Limited Memory:** For a normal laptop or desktop, memory (RAM) is often limited to 16 or 32 GB. So, it's kind of impossible to load all the data files together.\n2. **Limited CPU:** Libraries like pandas or numpy can utilize only 1 CPU at any point of time. As a result, even though the laptop/desktop/VM have multiple cores, we can't use those.\n\n**Dask** is a framework designed to overcome these limitations:\n\n1. **Parallel Computing:** Dask enables parallel computation using multi-core CPUs.\n2. **Out of Core Computing:** In case, the size of the data is larger than the memory (RAM), dask doesn't load all the data in-memory at a time. In turn, it streams the data from the disk as and when needed.\n\nDask can scale on thousand-machine clusters to handle hundreds of terabytes of data. At the same time, it works efficiently on a single machine as well, enabling analysis of moderately large datasets (100GB+) on relatively low power laptops.\n\nIn this notebook, we are going to focus on using Dask in a single machine.\n\n#### **Note: Since the number of CPUs available in Kaggle Kernel is just 4, this notebook takes lot of time to get executed. If you have a VM or a laptop with higher number of CPUs, this notebook will take much lesser time and help you to fully understand the power of Dask**"},{"metadata":{},"cell_type":"markdown","source":"# Start a Dask Client\n\nBy initiating a Client here, we are making sure that all the cores available for this Kernel is used while doing any computation."},{"metadata":{"trusted":true},"cell_type":"code","source":"client = Client(n_workers=3)\nclient","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This has created a **local** Dask Cluster utilizing all the 4 Cores. "},{"metadata":{},"cell_type":"markdown","source":"# Read train & submission data"},{"metadata":{"trusted":true},"cell_type":"code","source":"# Location of the parent directory for data\nDATA_DIR = \"/kaggle/input/predict-volcanic-eruptions-ingv-oe\"\n\n# Read train and submission data\ntrain_df = pd.read_csv(f\"{DATA_DIR}/train.csv\")\nsubmission_df = pd.read_csv(f\"{DATA_DIR}/sample_submission.csv\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Check the number of segments present in the training and submission files"},{"metadata":{"trusted":true},"cell_type":"code","source":"# Check the number of segments present in the training and submission files\ntrain_df.shape, submission_df.shape","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"There are 4431 segments for training and 4520 segments for testing"},{"metadata":{},"cell_type":"markdown","source":"# Explore train and test data files related to different segments"},{"metadata":{"trusted":true,"_kg_hide-input":true},"cell_type":"code","source":"# Define the datatypes for different sensor data\ndata_types = {\"sensor_1\" : np.float32, \n                 \"sensor_2\" : np.float32, \n                 \"sensor_3\" : np.float32,\n                 \"sensor_4\" : np.float32,\n                 \"sensor_5\" : np.float32,\n                 \"sensor_6\" : np.float32,\n                 \"sensor_7\" : np.float32,\n                 \"sensor_8\" : np.float32,\n                 \"sensor_9\" : np.float32,\n                 \"sensor_10\" : np.float32}","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Read data across different training segments\n\nTo read the CSV files we are going to use Dask. \n\n#### Dask doesn't read all the files to memory immediately. Instead, it creates a reference and reads the CSV files only when there is a need (for any computaion)"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ndd_train_seg = dd.read_csv(\n    urlpath=f\"{DATA_DIR}/train/*.csv\",\n    blocksize=None,\n    dtype=data_types, \n    include_path_column=\"segment_id\")\n\ndd_train_seg.head(2)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This process took just 4.24 seconds.\n\nIf we wanted to do the same process using pandas, we had to loop through all the files, read and load those one by one. That may take multiple minutes of time.\n\nHowever, the column defined as `segment_id` holds the absolute path of individual CSV files. For our purpose, we just need the segment_id. Hence, we will retain just the segment_id in the next step."},{"metadata":{"trusted":true},"cell_type":"code","source":"# Just keep the segment ID\ndd_train_seg.segment_id = dd_train_seg.segment_id.str.replace(\"/kaggle/input/predict-volcanic-eruptions-ingv-oe/train/\", \"\")\ndd_train_seg.segment_id = dd_train_seg.segment_id.str.replace(\".csv\", \"\")\n# Convert the segment ID column from String to int64\ndd_train_seg.segment_id = dd_train_seg.segment_id.astype(\"int64\")\n\ndd_train_seg.head()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This Dask DataFrame actually is a concatination of 4432 pandas DataFrame. Each Pandas DataFrame represents data for one of the segments and is identified by the value of the `segment_id`\n\n#### Let's verify. The resulting number of segments present in `dd_train_seg` should match with the segments present in the `train.csv`"},{"metadata":{},"cell_type":"markdown","source":"### Count the number of segments present in the Dask DataFrame."},{"metadata":{"trusted":true},"cell_type":"code","source":"%time\n# How many segments are there in training data?\nprint(\"Number of segments in dd_train_seg: \", dd_train_seg.segment_id.nunique().compute())","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This process took 10 micro seconds.\n\nThis is the first time, Dask reads all the train CSV files from the disk (using 4 workers), check the number of unique segments across the files\n\nAs we can see the number of segments present in this Dask DataFrame (4432) matches with the segments present in the train.csv"},{"metadata":{},"cell_type":"markdown","source":"### Shape of the Dask DataFrame\n\n#### How many columns are there in the Dask DF?"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nlen(dd_train_seg.columns)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### How many rows are there in the Dask DF?"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\nlen(dd_train_seg)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"**There are approximately 265 million rows. This is huge.**\n\nHere, the compute() function has been invoked implicitly by Dask. As expected, Dask needs to read all the CSV files from the disk, calculate the number or rows of each file and then add those values to compute the final result. It took arond 8 min 47 seconds. If we can increase the number of cores, this value will come down drastically."},{"metadata":{},"cell_type":"markdown","source":"### How big is the Dask DF if loaded in memory?"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ntrain_memory_usage = dd_train_seg.memory_usage(deep=True).compute()\n\nprint(\"Size (in GB) of the training segment DD (including index) : \", train_memory_usage.sum()/(10**9))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Size of the Dask DF is 12.76 GB. This is less than the size on the disk (15 GB) because while loading the CSV files the data types have been changed."},{"metadata":{},"cell_type":"markdown","source":"### Compute number of observations for every segment in the Dask DataFrame"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n_size = dd_train_seg.groupby(\"segment_id\").size().compute()\nprint(f\"In training DD, Length of the individual DataFrames for each segment {_size.unique()}\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Check if there are missing values for different sensors across different training segments"},{"metadata":{"trusted":true},"cell_type":"code","source":"# Helper function to make necessary computation on pandas dataframe related to individual segments. \ndef get_missing_sensors(df, sensor_names):\n    \"\"\"\n    Returns a DataFrame consisting of segment id, number of columns (sensors)\n    and percentage of missing data per sensor\n    \n    sensor_names: A list consisting of column names related to sensors\n    \n    \"\"\"\n    # Get the segment_id\n    segment_id = df.segment_id.unique()[0]\n    # Get the percentage of missing data across sensors\n    df_missing_percentage = df[sensor_names].isna().mean().to_frame().transpose()\n    df_missing_percentage[sensor_names] = df_missing_percentage[sensor_names].astype(np.float16)\n        \n    df_missing_percentage[\"segment_id\"] = segment_id\n    return df_missing_percentage","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n# Get the column names related to sensors\nsensor_names = [name for name in dd_train_seg.columns if \"sensor\" in name]\ndf_train_seg_missing = dd_train_seg.map_partitions(get_missing_sensors, sensor_names=sensor_names).compute()\n\ndf_train_seg_missing.head()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Columns with name `sensor_x` represents the percentage of missingness for that particular sensor for a segment. A value of `1` represents that particular sensor data is not present for the segment."},{"metadata":{},"cell_type":"markdown","source":"### Check if a sensor is completely missing for a particular segment?"},{"metadata":{"trusted":true},"cell_type":"code","source":"df_missingness_across_sensors = df_train_seg_missing[sensor_names].eq(1).sum()\n\n# Plot the missingness\ndf_missingness_across_sensors.plot.bar(figsize=(10, 6), title=\"Missing Sensors Across Training Segments\")\nplt.ylabel(\"Number of Segments with Missing Sensors\")\nplt.xlabel(\"Name of the Sensor\")\nplt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Get the groups of sensors which are missing across segments"},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_missing_groups(row):\n    row_value = row.values.tolist()\n    if row_value:\n        return \"_\".join(row_value)\n\nmissing_sensor_groups = df_train_seg_missing[sensor_names].apply(lambda row: row[row == 1].index, axis=1)\nmissing_sensor_groups_count = missing_sensor_groups.apply(lambda row: get_missing_groups(row)).dropna().value_counts()\n\nmissing_sensor_groups_count.plot.bar(figsize=(10, 6), title=\"Missing Sensors Across Training Segments\")\nplt.ylabel(\"Number of Segments with Missing Sensor Groups\")\nplt.xlabel(\"Name of the Sensor Groups\")\nplt.show()","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}