Tips on Efficient Coding
SAFE Research Data Center
Errikos Melissinos, Research Assistant - May 2022
|
Tips on Efficient Coding SAFE Research Data Center Errikos Melissinos, Research Assistant - May 2022 |
|
| Documentation: | |
|---|---|
| *.ipynb file | Please note that .ipynb file requires a Jupyter environment, for ex. Google Colab, Jupyter lab, |
| YouTube video (22 mins). Sections: | |
The cluster offers storage, cores, memory and convenience.
Before going into the multiprocessing I would like to look a little bit into a few tips for efficient programming.
So let's assume that we want to make a calculation for each row in a dataset:
import pandas as pd
import numpy as np
df0 = pd.read_csv("files/infl_unemp.csv").dropna().reset_index(drop=True)
df = df0.dropna()
countries = ['BE', 'BG', 'CY', 'CZ', 'DE', 'DK', 'EA19', 'EE', 'EL', 'ES',
'EU27_2020', 'FI', 'FR', 'HR', 'HU', 'IE', 'IS', 'IT', 'LT', 'LU',
'LV', 'MT', 'NL', 'PT', 'RO', 'SE', 'SI', 'SK', 'AT', 'NO', 'PL',
'CH', 'TR', 'US', 'UK']
weights = np.random.rand(len(countries))
weight_dict = dict(zip(countries,weights/sum(weights)))
weights_frame = pd.DataFrame(weights,columns=["weights"])
weights_frame["country"] = countries
df = pd.merge(df, weights_frame, on="country", how="left")
df
| Unnamed: 0 | country | date | year | month | hicp | unemp | first_year | inflation | lag_inflation | weights | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 48 | IE | 2022M02 | 2022 | 2 | 107.10 | 4.8 | 1996 | -2.20 | -1.00 | 0.235249 |
| 1 | 64 | AT | 2022M01 | 2022 | 1 | 113.93 | 4.8 | 1996 | -1.47 | -2.60 | 0.343967 |
| 2 | 65 | BE | 2022M01 | 2022 | 1 | 116.99 | 5.6 | 1996 | -2.50 | -0.79 | 0.269843 |
| 3 | 66 | BG | 2022M01 | 2022 | 1 | 114.80 | 4.5 | 2000 | -1.36 | -2.43 | 0.593467 |
| 4 | 67 | CY | 2022M01 | 2022 | 1 | 103.33 | 6.5 | 2000 | -0.89 | -1.91 | 0.734424 |
| ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 10052 | 10122 | NO | 1996M01 | 1996 | 1 | 69.20 | 4.8 | 1996 | -0.10 | -0.20 | 0.358546 |
| 10053 | 10123 | PT | 1996M01 | 1996 | 1 | 65.76 | 8.0 | 1996 | -0.33 | -0.14 | 0.110814 |
| 10054 | 10124 | SE | 1996M01 | 1996 | 1 | 75.41 | 8.7 | 1996 | -0.19 | -0.54 | 0.820810 |
| 10055 | 10125 | SI | 1996M01 | 1996 | 1 | 43.88 | 7.2 | 1996 | -0.55 | -0.64 | 0.947574 |
| 10056 | 10126 | UK | 1996M01 | 1996 | 1 | 67.80 | 8.2 | 1996 | -0.30 | -0.30 | 0.300982 |
10057 rows × 11 columns
# I want to perform the following operation on each row:
def operation(row):
return row["weights"]*(1.5*row["inflation"]+0.3*row["lag_inflation"]+0.2*row["unemp"])
# There is a lot of different ways to do this and I want to time my results for each option
from qgrid import show_grid
# this only works on jupyter hub, magic commands
%load_ext line_profiler
The line_profiler extension is already loaded. To reload it, use: %reload_ext line_profiler
df
| Unnamed: 0 | country | date | year | month | hicp | unemp | first_year | inflation | lag_inflation | weights | |
|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 48 | IE | 2022M02 | 2022 | 2 | 107.10 | 4.8 | 1996 | -2.20 | -1.00 | 0.235249 |
| 1 | 64 | AT | 2022M01 | 2022 | 1 | 113.93 | 4.8 | 1996 | -1.47 | -2.60 | 0.343967 |
| 2 | 65 | BE | 2022M01 | 2022 | 1 | 116.99 | 5.6 | 1996 | -2.50 | -0.79 | 0.269843 |
| 3 | 66 | BG | 2022M01 | 2022 | 1 | 114.80 | 4.5 | 2000 | -1.36 | -2.43 | 0.593467 |
| 4 | 67 | CY | 2022M01 | 2022 | 1 | 103.33 | 6.5 | 2000 | -0.89 | -1.91 | 0.734424 |
| ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 10052 | 10122 | NO | 1996M01 | 1996 | 1 | 69.20 | 4.8 | 1996 | -0.10 | -0.20 | 0.358546 |
| 10053 | 10123 | PT | 1996M01 | 1996 | 1 | 65.76 | 8.0 | 1996 | -0.33 | -0.14 | 0.110814 |
| 10054 | 10124 | SE | 1996M01 | 1996 | 1 | 75.41 | 8.7 | 1996 | -0.19 | -0.54 | 0.820810 |
| 10055 | 10125 | SI | 1996M01 | 1996 | 1 | 43.88 | 7.2 | 1996 | -0.55 | -0.64 | 0.947574 |
| 10056 | 10126 | UK | 1996M01 | 1996 | 1 | 67.80 | 8.2 | 1996 | -0.30 | -0.30 | 0.300982 |
10057 rows × 11 columns
%%timeit
# crude looping
res = []
for i in range(len(df)):
res.append(operation(df.iloc[i,:]))
df["int"]=res
1.39 s ± 14.4 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
%%timeit
# loop with iterrows
res = []
for _, row in df.iterrows():
res.append(operation(row))
df["int"]=res
677 ms ± 7.35 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
%%timeit
# loop using apply
df.apply(operation, axis=1)
195 ms ± 2.28 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)
# analyze the process
%lprun -f operation df.apply(operation, axis=1) # this is still going through the loop
Timer unit: 1e-06 s
Total time: 0.436895 s
File: /tmp/ipykernel_796523/3911977210.py
Function: operation at line 2
Line # Hits Time Per Hit % Time Line Contents
==============================================================
2 def operation(row):
3 10057 436895.0 43.4 100.0 return row["weights"]*(1.5*row["inflation"]+0.3*row["lag_inflation"]+0.2*row["unemp"])
df
| Unnamed: 0 | country | date | year | month | hicp | unemp | first_year | inflation | lag_inflation | weights | int | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 48 | IE | 2022M02 | 2022 | 2 | 107.10 | 4.8 | 1996 | -2.20 | -1.00 | 0.235249 | -0.621058 |
| 1 | 64 | AT | 2022M01 | 2022 | 1 | 113.93 | 4.8 | 1996 | -1.47 | -2.60 | 0.343967 | -0.696534 |
| 2 | 65 | BE | 2022M01 | 2022 | 1 | 116.99 | 5.6 | 1996 | -2.50 | -0.79 | 0.269843 | -0.773641 |
| 3 | 66 | BG | 2022M01 | 2022 | 1 | 114.80 | 4.5 | 2000 | -1.36 | -2.43 | 0.593467 | -1.109191 |
| 4 | 67 | CY | 2022M01 | 2022 | 1 | 103.33 | 6.5 | 2000 | -0.89 | -1.91 | 0.734424 | -0.446530 |
| ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 10052 | 10122 | NO | 1996M01 | 1996 | 1 | 69.20 | 4.8 | 1996 | -0.10 | -0.20 | 0.358546 | 0.268909 |
| 10053 | 10123 | PT | 1996M01 | 1996 | 1 | 65.76 | 8.0 | 1996 | -0.33 | -0.14 | 0.110814 | 0.117795 |
| 10054 | 10124 | SE | 1996M01 | 1996 | 1 | 75.41 | 8.7 | 1996 | -0.19 | -0.54 | 0.820810 | 1.061307 |
| 10055 | 10125 | SI | 1996M01 | 1996 | 1 | 43.88 | 7.2 | 1996 | -0.55 | -0.64 | 0.947574 | 0.400824 |
| 10056 | 10126 | UK | 1996M01 | 1996 | 1 | 67.80 | 8.2 | 1996 | -0.30 | -0.30 | 0.300982 | 0.331080 |
10057 rows × 12 columns
%%timeit
# use vectorised pandas processes
df["int"] = operation(df)
584 µs ± 14.6 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
%lprun -f operation operation(df) # this is only running once
Timer unit: 1e-06 s
Total time: 0.0019 s
File: /tmp/ipykernel_796523/3911977210.py
Function: operation at line 2
Line # Hits Time Per Hit % Time Line Contents
==============================================================
2 def operation(row):
3 1 1900.0 1900.0 100.0 return row["weights"]*(1.5*row["inflation"]+0.3*row["lag_inflation"]+0.2*row["unemp"])
%%timeit
# another way to do the same thing
df["int"] = df["weights"]*(1.5*df["inflation"]+0.3*df["lag_inflation"]+0.2*df["unemp"])
584 µs ± 13.9 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
%%timeit
# and if we need a lot of speed we go to numpy arrays
df["int"] = df["weights"].values*(1.5*df["inflation"].values+0.3*df["lag_inflation"].values+0.2*df["unemp"].values)
162 µs ± 1.51 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
df
| Unnamed: 0 | country | date | year | month | hicp | unemp | first_year | inflation | lag_inflation | weights | int | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 48 | IE | 2022M02 | 2022 | 2 | 107.10 | 4.8 | 1996 | -2.20 | -1.00 | 0.235249 | -0.621058 |
| 1 | 64 | AT | 2022M01 | 2022 | 1 | 113.93 | 4.8 | 1996 | -1.47 | -2.60 | 0.343967 | -0.696534 |
| 2 | 65 | BE | 2022M01 | 2022 | 1 | 116.99 | 5.6 | 1996 | -2.50 | -0.79 | 0.269843 | -0.773641 |
| 3 | 66 | BG | 2022M01 | 2022 | 1 | 114.80 | 4.5 | 2000 | -1.36 | -2.43 | 0.593467 | -1.109191 |
| 4 | 67 | CY | 2022M01 | 2022 | 1 | 103.33 | 6.5 | 2000 | -0.89 | -1.91 | 0.734424 | -0.446530 |
| ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 10052 | 10122 | NO | 1996M01 | 1996 | 1 | 69.20 | 4.8 | 1996 | -0.10 | -0.20 | 0.358546 | 0.268909 |
| 10053 | 10123 | PT | 1996M01 | 1996 | 1 | 65.76 | 8.0 | 1996 | -0.33 | -0.14 | 0.110814 | 0.117795 |
| 10054 | 10124 | SE | 1996M01 | 1996 | 1 | 75.41 | 8.7 | 1996 | -0.19 | -0.54 | 0.820810 | 1.061307 |
| 10055 | 10125 | SI | 1996M01 | 1996 | 1 | 43.88 | 7.2 | 1996 | -0.55 | -0.64 | 0.947574 | 0.400824 |
| 10056 | 10126 | UK | 1996M01 | 1996 | 1 | 67.80 | 8.2 | 1996 | -0.30 | -0.30 | 0.300982 | 0.331080 |
10057 rows × 12 columns
The standard library used in python is multiprocessing. However, I will jump directly to something that I find more convenient, namely concurrent.futures.
# import multiprocessing
import concurrent.futures
%%timeit
chunk_size = 1000
with concurrent.futures.ProcessPoolExecutor() as executor:
fs = [executor.submit(operation, df.iloc[x:x+chunk_size,:]) for x in range(0,df.shape[0],chunk_size) ]
results = [f.result() for f in fs]
df["int"]=pd.concat(results)
1e+03 ms ± 96.4 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
df
So, it does not really make sense to introduce parallel computation for relatively simple tasks. It costs in coding and the computational overveads can actually make it slower.
# def reg_unemp_on_lag_inf(frame):
# "This gives the slop coefficient of a regression of unemployment on lag_inflation"
# # print("bob")
# # # print(np.dot(frame["unemp"], frame["lag_inflation"]))
# # print(np.dot(frame["unemp"], frame["lag_inflation"])/np.dot(frame["lag_inflation"], frame["lag_inflation"]))
# # assert False
# return np.dot(frame["unemp"], frame["lag_inflation"].T)/np.dot(frame["lag_inflation"], frame["lag_inflation"])
import statsmodels.api as sm
def get_reg_results(frame):
temp = frame.copy(deep=False)
temp["const"] = 1
model = sm.OLS(temp["unemp"], temp[["const","inflation","lag_inflation"]])
results = model.fit()
for _ in range(20):
results = model.fit()
return results.params
def get_reg_results2(frame):
results = [[],[],[]]
for _ in range(100):
temp = frame.sample(frac=2/3)
# temp = temp.sample(int(df.shape[0]*2/3))
temp["const"] = 1
# print(frame)
# print("lala")
# print(temp)
model = sm.OLS(temp["unemp"], temp[["const","inflation","lag_inflation"]])
result = model.fit()
results[0].append(result.params[0])
results[1].append(result.params[1])
# results[2].append(result.params[2])
return pd.Series([(sum(series)/len(series) if len(series) != 0 else None) for series in results] ,index=["const","inflation","lag_inflation"])
get_reg_results2(df)# df
const 8.407389 inflation 0.115055 lag_inflation NaN dtype: float64
reg_unemp_on_lag_inf(df)
df.shape[0]
%%timeit
chunk_size = 1000
with concurrent.futures.ProcessPoolExecutor() as executor:
countries = df.loc[:,"country"].unique()
fs = [executor.submit(get_reg_results2, df.loc[df.loc[:,"country"]==country]) for country in countries]
results = np.column_stack((countries,np.array([f.result() for f in fs])))
temp = pd.DataFrame(results, columns=["country", "c_const","c_infl","c_lag_infl"])
df_temp = pd.merge(df, temp, on="country", how="left")
1.33 s ± 63.6 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
# %%timeit
# chunk_size = 1000
# with concurrent.futures.ProcessPoolExecutor() as executor:
# countries = df.loc[:,"country"].unique()
# fs = [executor.submit(reg_unemp_on_lag_inf, df.loc[df.loc[:,"country"]==country]) for country in df.loc[:,"country"].unique()]
# results = np.column_stack((countries,np.array([f.result() for f in fs])))
# temp = pd.DataFrame(results, columns=["country", "beta"])
# df_temp = pd.merge(df, temp, on="country", how="left")
df
# %%timeit
# df_temp = pd.merge(df, df.groupby("country").apply(reg_unemp_on_lag_inf).rename("beta").to_frame(), on="country",how="left")
%%timeit
df_temp = pd.merge(df, df.groupby("country").apply(get_reg_results).rename(columns={"const":"c_const","inflation":"c_infl","lag_inflation":"c_lag_infl"}), on="country",how="left")
df_temp
58.4 ms ± 1.19 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)
df
There is not much that I can actaully show you here as the title actually says it all.
However, even if the cluster offers a lot of memory it would also be good to know some practises in order to economise on the size of the data in memory. So, let me load a relatively large dataset:
# how to find the files that I am interested in:
# cd /SAFE_NAS