An implementation of Map/Reduce in Erlang (with examples)

map_reduce.erl

-module(map_reduce).
-compile(export_all).

% Buckets = tuple that will be sent to the map function
% Map_Func = map function.
%		Format: fun(Bucket, Reduce_PID), messages Reduce_PID with array of results.
% Reduce_Func = reduce function
%		Format: fun(Num_Buckets, Acc, [Results], Main_PID)
%		Receives message from Reduce_PID. 
%		Responsible for messaging Main_PID with results.

start(Module, Buckets, Map_Func, Reduce_Func) ->
	Reduce_PID = spawn(Module, Reduce_Func, [length(Buckets), 0, [], self()]),
	lists:foreach(fun(Bucket) -> spawn(Module, Map_Func, [Bucket, Reduce_PID]) end, Buckets),

	receive
		Results ->
			Results
	end.

example() ->
	% Calculate pythagorean triples whose sum is 1000
	% A^2 + B^2 = C^2
	% (Problem taken from Project Euler #6)
	List = lists:seq(1, 1000),
	Buckets = [{Min, Min+99, List} || Min <- lists:seq(1, 901, 100)],
	start(?MODULE, Buckets, example_map_function, example_reduce_function).

example_map_function({Min, Max, List}, Reduce_PID) ->
	link(Reduce_PID), % If the reduce process exits early, extra map processes will also end.
	Result = [A * B * C || 	A <- List,
				B <- List,
				C <- lists:seq(Min, Max),
				A*A + B*B == C*C andalso A + B + C == 1000],
	Reduce_PID ! Result.

example_reduce_function(N, Acc, Results, Main_PID) when Acc < N ->
	receive
		[] ->
			example_reduce_function(N, Acc+1, Results, Main_PID);
		Result ->
			% Based on the problem constraints, there will only ever be 1 result
			Main_PID ! lists:append(Results, Result),
			exit(done_early) 
	end;
example_reduce_function(_, _, Results, Main_PID) ->
	Main_PID ! Results.

example_without_map_reduce() ->
	Result = [A * B * C || 	A <- lists:seq(1, 1000),
				B <- lists:seq(1, 1000),
				C <- lists:seq(1, 1000),
				A*A + B*B == C*C andalso A + B + C == 1000],
	Result.