Pandas Data Manipulation and Analysis Questions

import pandas as pd
from datetime import datetime, timedelta
import pyarrow.parquet as pq
import pyarrow as pa
import os

# control metadata (persisted separately)
last_max_ts = pd.Timestamp("2025-11-23T00:00:00")  # read from metadata store
now = pd.Timestamp(datetime.utcnow())

# extract incremental rows from source (example: read SQL into df_new)
# df_new = query_sql(f"SELECT * FROM events WHERE event_ts > '{last_max_ts}'")
# here we mock:
df_new = pd.DataFrame({
    "user_id":[1,2,1],
    "event_ts":[now - pd.Timedelta(days=3), now - pd.Timedelta(days=2), now - pd.Timedelta(days=1)],
    "revenue":[10,20,5],
    "country":["US","US","CA"]
})
if df_new.empty:
    print("No new data")
else:
    # enrich with week key
    df_new['week_start'] = df_new['event_ts'].dt.to_period('W').apply(lambda p: p.start_time)
    # compute incremental aggregates
    agg_new = df_new.groupby(['week_start','country']).agg(
        users=('user_id','nunique'),
        total_revenue=('revenue','sum'),
        events=('user_id','count')
    ).reset_index()
    # load existing aggregates (if exists)
    out_path = "/mnt/curated/kpi_weekly.parquet"
    if os.path.exists(out_path):
        existing = pd.read_parquet(out_path)
        # drop rows for weeks we will upsert (idempotent)
        keep = existing[~existing['week_start'].isin(agg_new['week_start'].unique())]
        merged = pd.concat([keep, agg_new], ignore_index=True)
    else:
        merged = agg_new
    # write back atomically (write tmp then rename)
    tmp = out_path + ".tmp"
    table = pa.Table.from_pandas(merged)
    pq.write_table(table, tmp)
    os.replace(tmp, out_path)
    # update metadata last_max_ts to max event_ts processed
    new_last = df_new['event_ts'].max()
    # persist new_last to metadata store
    print(f"Updated aggregates and metadata to {new_last}")

import pandas as pd

# assume orders is the input DataFrame with columns:
# ['order_id','customer_id','order_date','amount','quantity']

customer_metrics = (
    orders
    .groupby('customer_id')
    .agg(
        total_orders=pd.NamedAgg(column='order_id', aggfunc='count'),
        total_amount=pd.NamedAgg(column='amount', aggfunc='sum'),
        avg_order_value=pd.NamedAgg(column='amount', aggfunc='mean'),
        total_items=pd.NamedAgg(column='quantity', aggfunc='sum')
    )
    .reset_index()
)

# optional: format numeric columns, handle NaNs
customer_metrics['avg_order_value'] = customer_metrics['avg_order_value'].fillna(0)

import pandas as pd

def top10_customers(df):
    # drop rows with missing customer_id (or alternatively fillna('UNKNOWN'))
    df = df.dropna(subset=['customer_id'])
    # aggregate total spend per customer
    totals = df.groupby('customer_id', as_index=False)['amount'].sum()
    # return top 10 by total (deterministic tie-breaker by customer_id)
    return totals.sort_values(['amount', 'customer_id'], ascending=[False, True]).head(10)

def top10_with_ties(df):
    df = df.dropna(subset=['customer_id'])
    totals = df.groupby('customer_id', as_index=False)['amount'].sum()
    # determine cutoff = 10th largest total
    cutoff = totals['amount'].nlargest(10).min()
    return totals[totals['amount'] >= cutoff].sort_values(['amount','customer_id'], ascending=[False,True])

import pandas as pd

# example Series
s = pd.Series(['A', 'B', 'A', None, 'C', 'B', 'A', 'D', None, 'E', 'B'])

# Absolute counts including NaNs
abs_counts = s.value_counts(dropna=False)
print("Absolute counts (including NaN):")
print(abs_counts)

# Relative frequencies (normalized) including NaNs
rel_freq = s.value_counts(dropna=False, normalize=True)
print("\nRelative frequencies (including NaN):")
print(rel_freq)

# Top 5 categories by absolute count with their normalized frequencies
top5 = abs_counts.head(5).to_frame(name='count')
top5['freq'] = rel_freq.loc[top5.index]
print("\nTop 5 (count + normalized frequency):")
print(top5)

import pandas as pd

# Pre-tune dtypes for both tables to reduce memory (example)
dim_dtypes = {'user_id': 'int64', 'country': 'category', 'signup_date': 'datetime64[ns]'}
fact_dtypes = {'user_id': 'int64', 'event_time': 'datetime64[ns]', 'event_type': 'category'}

# Load dimension into memory and set index for fast lookup
dim = pd.read_csv('dimension.csv', dtype=dim_dtypes, parse_dates=['signup_date'])
dim.set_index('user_id', inplace=True)

chunksize = 2_000_000  # adjust based on available RAM
out_path = 'joined.parquet'
first_write = True

for chunk in pd.read_csv('fact.csv', dtype=fact_dtypes, parse_dates=['event_time'], chunksize=chunksize):
    # Option 1: merge using join on index (faster if dim is indexed)
    chunk = chunk.set_index('user_id', drop=False)
    merged = chunk.join(dim, on='user_id', how='left', rsuffix='_dim')
    merged.reset_index(drop=True, inplace=True)

    # Optional: free memory by downcasting numeric cols or converting to categories
    merged.to_parquet(out_path, engine='pyarrow', compression='snappy', index=False,
                      append=not first_write)
    first_write = False
    del chunk, merged