[FEAT] user definable reduce aggregation

[samster25]

Allow users to run their own two stage reduce.

something like:

df.reduce(per_partition_reduce_fn, global_reduce_fn)

or if they are the same

df.reduce(both_reduce_fn)