agrègent un dataframe Dask et produire une trame de données d'agrégats

Question

J'ai un dataframe Dask qui ressemble à ceci:

url  referrer session_id ts     customer 
url1 ref1  xxx  2017-09-15 00:00:00 a.com 
url2 ref2  yyy  2017-09-15 00:00:00 a.com 
url2 ref3  yyy  2017-09-15 00:00:00 a.com 
url1 ref1  xxx  2017-09-15 01:00:00 a.com 
url2 ref2  yyy  2017-09-15 01:00:00 a.com 

Je veux regrouper les données sur l'URL et l'horodatage, les valeurs de colonnes globales et produire une trame de données qui serait ressemble à ceci:

customer url ts     page_views visitors referrers 
a.com url1 2017-09-15 00:00:00 1   1  [ref1] 
a.com url2 2017-09-15 00:00:00 2   2  [ref2, ref3] 

Dans SQL Spark, je peux le faire comme suit:

select 
    customer, 
    url, 
    ts, 
    count(*) as page_views, 
    count(distinct(session_id)) as visitors, 
    collect_list(referrer) as referrers 
from df 
group by customer, url, ts 

Y a-t-il un moyen de le faire avec les dataframes Dask? J'ai essayé, mais je ne peux calculer les colonnes agrégées séparément, comme suit:

# group on timestamp (rounded) and url 
grouped = df.groupby(['ts', 'url']) 

# calculate page views (count rows in each group) 
page_views = grouped.size() 

# collect a list of referrer strings per group 
referrers = grouped['referrer'].apply(list, meta=('referrers', 'f8')) 

# count unique visitors (session ids) 
visitors = grouped['session_id'].count() 

Mais je ne peux pas sembler trouver une bonne façon de produire une trame de données combinée que j'ai besoin.

Answer 1

Ce qui suit ne fait œuvre:

gb = df.groupby(['customer', 'url', 'ts']) 
gb.apply(lambda d: pd.DataFrame({'views': len(d), 
    'visitiors': d.session_id.count(), 
    'referrers': [d.referer.tolist()]})).reset_index() 

(en présumant les visiteurs doivent être uniques selon le sql ci-dessus) Vous pouvez définir la meta de la sortie.