Only because the question is well written and it seemed like a nice puzzle, here's some magic.
Potentially you'll have to store a lot of data, so you need to compress the frame as much as possible and do several passes through the base. If the database contains not primitive objects, convert those into integers, if you do multiprocessing, the dataframe will be copied into subprocesses, so keeping it contents small helps.
The runtime depends on the length of the dataframe but also on the number of unique stores, unique products and the size of a chunk of pairs to count. Spreading the work to many subprocesses can speed up things but there is constant cost to all the functions which will accumulate. For example, pandas' own methods will run faster on a single ten thousand rows dataframe than on a dozen of thousand row frames. And when you're running nested calls on sub dataframes of unpredictable size things get complicated. You'll probably have to experiment a bit to find a chunksize with optimal speed\memory usage.
Test runtimes with smaller numbers first. Including less shops and products. That being said, this is not a quick task. On high end machine it completes in about ten minutes.
import pandas as pd, numpy as np
df = pd.DataFrame({
'store':np.random.randint(0,int(2e4),int(5e6)),
'product':np.random.randint(0,int(5e4),int(5e6))
}).sort_values('store')
products = df['product'].unique()
N, chunksize, Ntop = len(products), int(1e4), 200
dtype = np.min_scalar_type(max(products.max(),N))
df = df.astype(dtype)
def store_cats(df):
df = df.astype('category')
cats = [df[x].cat.categories for x in df.columns]
for col in df.columns:
df[col] = df[col].cat.codes
return df, cats
def restore_cats(summary,cats):
for col in ['product_x','product_y']:
summary[col] = pandas.Categorical.from_codes(summary[col], cats)
def subsets(n = chunksize):
n = int(n)
res = [frozenset(products[i:i+n]) for i in range(0,N,n)]
info = 'In total there will be {:.1E} pairs, per pass {:.1E} will be checked, thats up to around {} mb per pass, {} passes'
print(info.format((N**2),(n*N),(n*N*3*8/1e6),len(res)))
return res
def count(df,subset):
res = df.merge(df,on = 'store')\
.query('(product_x < product_y) and product_x in @subset')\
.groupby(['product_x','product_y'])\
.count()\
.astype(dtype)\
.reset_index()
return res
def one_pass(gr,subset):
per_group = gr.apply(count,subset)
total_counts = per_group.sort_values(['product_x','product_y'])\
.groupby(['product_x','product_y'])\
.agg('sum')\
.sort_values('store',ascending=False)[:Ntop]\
.copy().reset_index()
return total_counts
def merge_passes(dfs):
res = pd.concat(dfs,ignore_index=True)
res = res.append(res.rename(columns={'product_x':'product_y','product_y':'product_x'}),ignore_index=True)
res = res.sort_values('store',ascending=False)[:Ntop]
return res
from concurrent.futures import as_completed, ProcessPoolExecutor as Pool
gr = df.groupby('store',as_index = False)
def worker(subset):
return one_pass(gr,subset)
def run_progress(max_workers=2,chunksize=chunksize):
from tqdm.auto import tqdm
with Pool(max_workers = max_workers) as p:
futures = [p.submit(worker,subset) for subset in subsets(chunksize)]
summaries = [x.result() for x in tqdm(as_completed(futures),total=len(futures))]
return merge_passes(summaries)
