%% Introduction to |datastore|
% Creating a |datastore| allows you to access a collection of data in a
% chunk-based manner.  A |datastore| can process arbitrarily large amounts
% of data, and the data can even be spread across multiple files. You can
% create a |datastore| for a collection of tabular text files (demonstrated
% next), a SQL database (Database Toolbox(TM) required)
% or a Hadoop(R) Distributed File System (HDFS(TM)).
%
% Create a datastore for a collection of tabular text files and preview the
% contents.
ds = datastore('airlinesmall.csv');
dsPreview = preview(ds);
dsPreview(:,10:15)

%%
% The datastore automatically parses the input data and makes a best guess
% as to the type of data in each column.  In this case, use the
% |'TreatAsMissing'| Name-Value pair argument to have |datastore| replace
% the missing values correctly. For numeric variables (such as
% |'AirTime'|), |datastore| replaces every |'NA'| string with a |NaN|
% value, which is the IEEE arithmetic representation for Not-a-Number.
ds = datastore('airlinesmall.csv', 'TreatAsMissing', 'NA');
ds.SelectedFormats{strcmp(ds.SelectedVariableNames, 'TailNum')} = '%s';
ds.SelectedFormats{strcmp(ds.SelectedVariableNames, 'CancellationCode')} = '%s';
dsPreview = preview(ds);
dsPreview(:,{'AirTime','TaxiIn','TailNum','CancellationCode'})

%% Scan for rows of interest
% |datastore| objects contain an internal pointer to keep track of which
% chunk of data the |read| function returns next. Use the |hasdata| and
% |read| functions to step through the entire data set, and filter the data
% set to only the rows of interest. In this case, the rows of interest are
% flights on United Airlines ("UA") departing from Boston ("BOS").

subset = [];

while hasdata(ds)
    t = read(ds);
    t = t(strcmp(t.UniqueCarrier, 'UA') & strcmp(t.Origin, 'BOS'), :);
    subset = vertcat(subset, t);
end

subset(1:10,[9,10,15:17])

%% Introduction to |mapreduce|
% MapReduce is an algorithmic technique to "divide and conquer" big data
% problems.  In MATLAB,  |mapreduce| requires three input arguments:
%
% # A |datastore| to read data from
% # A "mapper" function that is given a subset of the data to operate on.
% The output of the map function is a partial calculation.  |mapreduce| calls
% the mapper function one time for each chunk in the |datastore|, with each
% call operating independently.
% # A "reducer" function that is given the aggregate outputs from the
% mapper function.  The reducer function finishes the computation begun by
% the mapper function, and outputs the final answer.
%
% This is an over-simplification to some extent, since the output of a call
% to the mapper function can be shuffled and combined in interesting ways
% before being passed to the reducer function.  This will be examined
% later in this example.

%% Use |mapreduce| to perform a computation
% A simple use of |mapreduce| is to find the longest flight time in the
% entire airline data set.  To do this:
%
% # The "mapper" function computes the maximum of each chunk from the
% |datastore|.
% # The "reducer" function then computes the maximum value among all of the
% maxima computed by the calls to the mapper function.
%
% First, reset the |datastore| and filter the variables to the one
% column of interest.

reset(ds);
ds.SelectedVariableNames = {'ActualElapsedTime'};

%%
% Write the mapper function, |maxTimeMapper.m|.  It takes
% three input arguments:
%
% # The input data, which is a table obtained by applying the |read|
% function to the |datastore|.
% # A collection of configuration and contextual information, |info|. This
% can be ignored in most cases, as it is here.
% # An intermediate data storage object, which records the results of the
% calculations from the mapper function. Use the |add| function to 
% add Key/Value pairs to this intermediate output. In this example, the
% name of the key (|'MaxElapsedTime'|) is arbitrary.
% 
% Save the following mapper function (|maxTimeMapper.m|) in
% your current folder.
type maxTimeMapper

%%
% Next, write the reducer function.  It also takes three input arguments:
%
% # A set of input "keys".  Keys will be discussed further below, but they
% can be ignored in some simple problems, as they are here.
% # An intermediate data input object that |mapreduce| passes to the reducer
% function. This data is in the form of Key/Value pairs, and you use the
% |hasnext| and |getnext| functions to iterate through the values for each
% key.
% # A final output data storage object. Use the |add| and |addmulti| functions to
% directly add Key/Value pairs to the output.
%
% Save the following reducer function (|maxTimeReducer.m|) in
% your current folder.
type maxTimeReducer

%%
% Once the mapper and reducer functions are written and saved in your
% current folder, you can call |mapreduce| using the |datastore|, mapper
% function, and reducer function.  If you have Parallel Computing Toolbox (PCT),
% MATLAB will automatically start a pool and parallelize execution.
% Use the |readall| function to display
% the results of the MapReduce algorithm.

result = mapreduce(ds, @maxTimeMapper, @maxTimeReducer);
readall(result)

%% Use of keys in |mapreduce|
% The use of keys is an important and powerful feature of |mapreduce|. Each
% call to the mapper function adds intermediate results to one or more
% named "buckets", called keys.  The number of calls to the mapper function
% by |mapreduce| corresponds to the number of chunks in the |datastore|.
%
% If the mapper function adds values to multiple keys, this leads to
% multiple calls to the reducer function, with each call working on only
% one key's intermediate values. The |mapreduce| function automatically
% manages this data movement between the map and reduce phases of the
% algorithm.
%
% This flexibility is useful in many contexts.  The example below uses keys
% in a relatively obvious way for illustrative purposes.

%% Calculating group-wise metrics with |mapreduce|
%
% The behavior of the mapper function in this application is more complex.
% For every flight carrier found in the input data, use the |add| function
% to add a vector of values.  This vector is a count of the number of
% flights for that carrier on each day in the 21+ years of data.  The
% carrier code is the key for this vector of values.  This ensures that all
% of the data for each carrier will be grouped together when |mapreduce|
% passes it to the reducer function.
%
% Save the following mapper function (|countFlightsMapper.m|) in your
% current folder.
type countFlightsMapper

%%
% The reducer function is less complex.  It simply iterates over the
% intermediate values and adds the vectors together.  At completion, it
% outputs the values in this aggregate vector.  Note that the reduce
% function does not need to sort or examine the |intermediateKeysIn| values;
% each call to the reduce function by |mapreduce| only passes the values
% for one airline carrier.
%
% Save the following reducer function (|countFlightsReducer.m|) in your
% current folder.
type countFlightsReducer

%%
% Reset the |datastore| and select the variables of interest. Once
% the mapper and reducer functions are written and saved in your current
% folder, you can call |mapreduce| using the |datastore|, mapper
% function, and reducer function.

reset(ds);
ds.SelectedVariableNames = {'Year', 'Month', 'DayofMonth', 'UniqueCarrier'};

result = mapreduce(ds, @countFlightsMapper, @countFlightsReducer);
result = readall(result);

%%
% In case this example was run with only the sample data set,
% load the results of the |mapreduce| algorithm run on the entire data set.
load airlineResults

%% Visualizing  the results
% Using only the top 7 carriers, apply a filter to the data to smooth out
% the effects of weekend travel.  This would otherwise clutter the
% visualization.

lines = result.Value;
lines = horzcat(lines{:});
[~,sortOrder] = sort(sum(lines), 'descend');
lines = lines(:,sortOrder(1:7));
result = result(sortOrder(1:7),:);

lines(lines==0) = nan;
for carrier=1:size(lines,2)
    lines(:,carrier) = filter(repmat(1/7, [7 1]), 1, lines(:,carrier));
end

%%
% Plot the data.
figure('Position',[1 1 800 600]);
plot(datetime(1987,10,1):caldays(1):datetime(2008,12,31),lines)
title ('Domestic airline flights per day per carrier')
xlabel('Date')
ylabel('Flights per day (7-day moving average)')
legend(result.Key, 'Location', 'South')

%%
% The plot shows the emergence of Southwest Airlines (WN) during this time
% period.

%% Learning more
% This example only scratches the surface of what is possible with
% |mapreduce|. See the documentation for |mapreduce| for more information,
% including information on using it with Hadoop and MATLAB Distributed Computing
% Server.
