<jupyter_start><jupyter_text>Rainfall Analysis - 525 Group 22  UsageFor easiest readability, please view the notebook as a webpage [here](https://ubc-mds.github.io/rainfall_group22/), or download and open the raw [html](https://github.com/UBC-MDS/rainfall_group22/blob/main/notebooks/milestone1.html).This notebook is to be run in the 525 conda environment. I experienced some issues with `r2py`, but was able to solve them by adding `python=3.8.6` to the yaml file and recreate the environment.We have used some helper functions, so before running please make sure to clone the repo: [https://github.com/UBC-MDS/rainfall_group22](https://github.com/UBC-MDS/rainfall_group22).Below we will install some extra dependencies, you will need to restart the kernel after installed the first time.<jupyter_code># Installations
import sys

# used for pretty printing summary table colours
%conda install --yes --prefix {sys.prefix} seaborn

# our package from 524, used to pretty-print file sizes during download. Not super necessary but fun to use.
%pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple nicenumber

# used to show file download progress bar
%pip install tqdm
# make sure root dir is added to sys.path for our imports from /src to work
from pathlib import Path
root_name = 'rainfall_group22'
p_root = Path.cwd()
print('p_root:', p_root)

if not p_root.name == root_name:
    p_root = p_root.parent
    sys.path.append(str(p_root))
    
assert p_root.name == root_name, 'Make sure notebook is run from cloned repo!'

from src import download as dl
from src import functions as f
from nicenumber import nicenumber as nn

import pandas as pd
import numpy as np
from collections import defaultdict

import rpy2.rinterface
import dask.dataframe as dd

# install the packages https://arrow.apache.org/docs/python/install.html
import pyarrow.dataset as ds
import pyarrow as pa
import pyarrow.parquet as pq

# How to install put instructions https://anaconda.org/conda-forge/rpy2
import rpy2.rinterface

# install this https://pypi.org/project/rpy2-arrow/#description  pip install rpy2-arrow
# have to install this as well conda install -c conda-forge r-arrow 
import rpy2_arrow.pyarrow_rarrow as pyra

import pyarrow.feather as feather
%load_ext rpy2.ipython
%load_ext memory_profiler<jupyter_output>The rpy2.ipython extension is already loaded. To reload it, use:
  %reload_ext rpy2.ipython
The memory_profiler extension is already loaded. To reload it, use:
  %reload_ext memory_profiler<jupyter_text>1. Download Data<jupyter_code>%%time
%%memit

# download and unzip data files
files = dl.download_files('data.zip', chunk_size=10)

if files:
    dl.unzip(p=files[0], p_dst='csv', delete=False)<jupyter_output>INFO    58   src.download               Downloading 814MB file in 10MB chunks.
100%|██████████| 814M/814M [03:43<00:00, 3.65MiB/s]
INFO    69   src.download               File downloaded to: /Users/Jayme/OneDrive/MDS/525/rainfall_group22/data/data.zip
INFO    126  src.download               Unpacking zip to: /Users/Jayme/OneDrive/MDS/525/rainfall_group22/data/csv<jupyter_text>2. Combine CSVs<jupyter_code># set download directories
p_data = dl.p_data # top level data dir
p_csv = p_data / 'csv' # sub dir for saving loose csvs
p_combined = p_data / 'rainfall.csv' # main csv file to use
%%time
%%memit

# combine csvs with pandas
csvs = [p for p in p_csv.glob('*.csv')]
dfs = []

# load individual dfs and save to list
for p in csvs:
    model_name = p.name.split('_')[0]

    df = pd.read_csv(p) \
        .assign(model=model_name)

    dfs.append(df)

# concat all dfs
df = pd.concat(dfs) \
    .rename(columns={'rain (mm/day)': 'rain'})<jupyter_output>peak memory: 12317.36 MiB, increment: 11777.62 MiB
CPU times: user 57.3 s, sys: 11.7 s, total: 1min 8s
Wall time: 1min 11s<jupyter_text>RuntimesTimes to combine dataframe csvs for each team member:|User|OS|Processor|RAM|Load Time (s)||:--|:--|:--|:--|--:||Jayme|Mac OS|2.4 GHz 8-Core Intel Core i9|32 GB 2667 MHz DDR4| 62||Zhiyong|Mac OS |2.6 GHz Dual-Core Intel Core i5 |8 GB 1600 MHz DDR3 |105|Marc|Windows 10 Education |Core(TM) i7-7700HQ 2.8 GHz |16 GB |86<jupyter_code>print(df.shape)
df.head()
df.info()
# save combined data back to csv
df.to_csv(p_combined, index=False)
# check size of full csv on disk
nn.to_human(p_combined.stat().st_size, prec=1, family='filesize')
%%time
%%memit
# Load csv with dask

ddf = dd.read_csv(p_combined, assume_missing=True)
ddf.head()<jupyter_output><empty_output><jupyter_text>3. Python EDAHere we will investigate and summarize the following approaches to reduce memory usage while performing a simple EDA (find maximum rainfall):1. Baseline2. Load data in chunks3. Load only columns of interest4. Dask  3.1 BaselineNaive approach, read all columns with pandas `read_csv`<jupyter_code>print_max_rain = lambda x: print(f'Max rainfall: {x:.2f} mm/day')
%%time
%%memit

max_rain_baseline = pd.read_csv(p_combined).rain.max()
print_max_rain(max_rain_baseline)<jupyter_output>Max rainfall: 432.94 mm/day
peak memory: 14904.34 MiB, increment: 3652.56 MiB
CPU times: user 1min, sys: 11.3 s, total: 1min 11s
Wall time: 1min 13s<jupyter_text>3.2 Load data in chunks<jupyter_code>%%time
%%memit

max_rain_chunks = np.finfo('float64').min

for df_chunk in pd.read_csv(p_combined, chunksize=1_000_000):
    cur_max = df_chunk.rain.max()
    if cur_max > max_rain_chunks:
        max_rain_chunks = cur_max

print_max_rain(max_rain_chunks)<jupyter_output>Max rainfall: 432.94 mm/day
peak memory: 9658.38 MiB, increment: 515.60 MiB
CPU times: user 55.9 s, sys: 2.77 s, total: 58.6 s
Wall time: 59.4 s<jupyter_text>3.3 Load only columns of interest<jupyter_code>%%time
%%memit

df_one_col = pd.read_csv(p_combined, usecols=['rain'])
max_rain_one = df_one_col.rain.max()
print_max_rain(max_rain_one)<jupyter_output>Max rainfall: 432.94 mm/day
peak memory: 9637.20 MiB, increment: 10.33 MiB
CPU times: user 27.6 s, sys: 1.27 s, total: 28.9 s
Wall time: 29.4 s<jupyter_text>3.4 Dask<jupyter_code>%%time
%%memit

ddf = dd.read_csv(p_combined)
max_rain_dask = ddf.rain.max().compute()

print_max_rain(max_rain_dask)<jupyter_output>Max rainfall: 432.94 mm/day
peak memory: 11811.69 MiB, increment: 2174.47 MiB
CPU times: user 1min 22s, sys: 14.5 s, total: 1min 37s
Wall time: 28.7 s<jupyter_text>SummaryThe following table summarizes memory usage and execution time for the simple EDA:<jupyter_code>m_results = dict(
    baseline=[14973, 65],
    chunks=[1323, 55],
    single_column=[5452, 27],
    dask=[5529, 25])

pd.DataFrame \
    .from_dict(
        m_results,
        orient='index',
        columns=['Peak Memory Usage (MB)', 'Execution Time (S)']) \
    .rename_axis('Method') \
    .style.pipe(f.bg, rev=False)<jupyter_output><empty_output><jupyter_text>Observations- To find the maximum rainfall, in this case we only needed one column in the table, therefor loading all columns was redundant.- Loading data in chunks reduced our execution time slightly, and greatly reduced peak memory usage.- Both loading only a single column and Dask had similar memory usage, with Dask executing slightly faster (25s).- Overall Dask reduced our memory usage by ~1/3 and execution time by ~60%.  4. R EDA  4.1 Save data to multiple formatsHere we will compare different methods to make the dataframe available in R:1. Feather2. Parquet<jupyter_code># load csv and create arrow table
dataset = ds.dataset(p_combined, format='csv')
arrow_table = dataset.to_table()
%%time
%%memit

# save feather file

p_feather = p_data / 'data.feather'
feather.write_feather(arrow_table, p_feather)
%%time
%%memit

# save parquet file
# NOTE parquet saves the file in a directory, need to get file path after

p_parquet_dir = p_data / 'parquet'
pq.write_to_dataset(arrow_table, p_parquet_dir)

# get parquet file path
p_parquet = list(p_parquet_dir.glob('*'))[0]
# check filesizes on disk
m_results = defaultdict(dict)
m_files = dict(
    feather=p_feather,
    parquet=p_parquet)

for name, p in m_files.items():
    size = p.stat().st_size
    size_human = nn.to_human(size, prec=1, family='filesize')

    m = m_results[name]
    m['size_disk'] = size

    print(f'{name}: {size_human}')<jupyter_output>feather: 1.1GB
parquet: 565.9MB<jupyter_text>4.2 Load data and perform EDA<jupyter_code>%%R

library(arrow)
library(dplyr)

# if run in jupyter need to bump root dir up one level from /rainfall_group22/notebooks to /rainfall_group22
cwd <- getwd()
if (grepl("notebook", cwd, fixed=TRUE)) {
    setwd("..")
}

# set filepaths in R
p_feather = "data/data.feather"
p_parquet = Sys.glob("data/parquet/*")

print(p_feather)
print(p_parquet)
%%time
%%R

# load feather
df_feather <- arrow::read_feather(p_feather)
%%time
%%R

# load parquet
df_parquet <- arrow::read_parquet(p_parquet)
%%R

print(class(df_feather))

max_rainfall <- df_feather[!is.na(df_feather$rain), ] %>% select('rain') %>% max()

print(max_rainfall)<jupyter_output>[1] "tbl_df"     "tbl"        "data.frame"
[1] 432.9395<jupyter_text>Summary<jupyter_code># compare results metrics

m_feather = dict(
    peak_memory=9417,
    time_write=5.82,
    time_load=12.4)

m_parquet = dict(
    peak_memory=12929,
    time_write=10.9,
    time_load=8.9)
    
m_results['feather'].update(m_feather)
m_results['parquet'].update(m_parquet)

pd.DataFrame \
    .from_dict(m_results).T \
    .rename_axis('Method') \
    .sort_index(axis=1) \
    .assign(time_total=lambda x: x.time_write + x.time_load) \
    .style \
    .pipe(f.bg, rev=False) \
    .format('{:,.0f}') \
    .format('{:.2f}', subset=['time_write', 'time_load', 'time_total'])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11494610281923714, 'fraction_numerical': 0.03783167495854063, 'mean_word_length': 4.571016166281755, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 43, 'https://': 9, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '291212', 'license': 'MIT', 'n_tokens_mistral': 3803, 'n_tokens_neox': 3457, 'n_words': 991, 'path': 'notebooks/milestone1.ipynb', 'repo_name': 'UBC-MDS/rainfall_group22', 'size': '9648'}