MapReduce

MapReduceは、クラスタ上で並列分散アルゴリズムを使用してビッグデータセットを処理および生成するためのプログラミングモデルおよび関連する実装です。^{[ 1 ] [ 2 ] [ 3 ]}

MapReduceプログラムは、フィルタリングとソート（例えば、学生を名字順にキューにソートし、名前ごとに1つのキューを作成するなど）を実行するマッププロシージャと、集計操作（例えば、各キュー内の学生数をカウントし、名前の出現頻度を算出するなど）を実行するリデュースメソッドで構成されます。「MapReduceシステム」（「インフラストラクチャ」または「フレームワーク」とも呼ばれる）は、分散サーバーを統合し、さまざまなタスクを並列実行し、システムのさまざまな部分間のすべての通信とデータ転送を管理し、冗長性と耐障害性を提供​​することで、処理を統括します。

このモデルは、データ分析のための分割適用結合戦略の特殊化です。^{[ 4 ]} これは、関数型プログラミングで一般的に使用されるmap 関数とreduce関数に触発されていますが、^{[ 5 ]} MapReduce フレームワークにおけるそれらの目的は、元の形式とは異なります。^{[ 6 ]} MapReduce フレームワークの主な貢献は、実際の map 関数と reduce 関数 (たとえば、1995 年のMessage Passing Interface標準の^{[ 7 ]} reduce ^{[ 8 ]}およびscatter ^{[ 9 ]}操作に似ています) ではなく、並列化によってさまざまなアプリケーションで実現されるスケーラビリティとフォールトトレランスです。そのため、 MapReduce のシングルスレッド実装は通常、従来の (MapReduce ではない) 実装よりも高速ではありません。利点は通常、マルチプロセッサ ハードウェア上のマルチスレッド実装でのみ見られます。 ^{[ 10 ]}このモデルの使用は、MapReduceフレームワークの最適化された分散シャッフル操作（ネットワーク通信コストを削減する）とフォールトトレランス機能が適用される場合にのみ有益です。通信コストの最適化は、優れたMapReduceアルゴリズムにとって不可欠です。^{[ 11 ]}

MapReduce libraries have been written in many programming languages, with different levels of optimization. A popular open-source implementation that has support for distributed shuffles is part of Apache Hadoop. The name MapReduce originally referred to the proprietary Google technology, but has since become a generic trademark. By 2014, Google was no longer using MapReduce as its primary big data processing model,^[12] and development on Apache Mahout had moved on to more capable and less disk-oriented mechanisms that incorporated full map and reduce capabilities.^[13]

MapReduce is a framework for processing parallelizable problems across large datasets using a large number of computers (nodes), collectively referred to as a cluster (if all nodes are on the same local network and use similar hardware) or a grid (if the nodes are shared across geographically and administratively distributed systems, and use more heterogeneous hardware). Processing can occur on data stored either in a filesystem (unstructured) or in a database (structured). MapReduce can take advantage of the locality of data, processing it near the place it is stored in order to minimize communication overhead.

A MapReduce framework (or system) is usually composed of three operations (or steps):

1. Map: each worker node applies the map function to the local data, and writes the output to a temporary storage. A master node ensures that only one copy of the redundant input data is processed.
2. Shuffle: worker nodes redistribute data based on the output keys (produced by the map function), such that all data belonging to one key is located on the same worker node.
3. Reduce: worker nodes now process each group of output data, per key, in parallel.

MapReduce allows for the distributed processing of the map and reduction operations. Maps can be performed in parallel, provided that each mapping operation is independent of the others; in practice, this is limited by the number of independent data sources and/or the number of CPUs near each source. Similarly, a set of 'reducers' can perform the reduction phase, provided that all outputs of the map operation that share the same key are presented to the same reducer at the same time, or that the reduction function is associative. While this process often appears inefficient compared to algorithms that are more sequential (because multiple instances of the reduction process must be run), MapReduce can be applied to significantly larger datasets than a single "commodity" server can handle – a large server farm can use MapReduce to sort a petabyte of data in only a few hours.^[14] The parallelism also offers some possibility of recovering from partial failure of servers or storage during the operation: if one mapper or reducer fails, the work can be rescheduled – assuming the input data are still available.

Another way to look at MapReduce is as a 5-step parallel and distributed computation:

1. Prepare the Map() input – the "MapReduce system" designates Map processors, assigns the input key K1 that each processor would work on, and provides that processor with all the input data associated with that key.
2. Run the user-provided Map() code – Map() is run exactly once for each K1 key, generating output organized by key K2.
3. "Shuffle" the Map output to the Reduce processors – the MapReduce system designates Reduce processors, assigns the K2 key each processor should work on, and provides that processor with all the Map-generated data associated with that key.
4. Run the user-provided Reduce() code – Reduce() is run exactly once for each K2 key produced by the Map step.
5. Produce the final output – the MapReduce system collects all the Reduce output, and sorts it by K2 to produce the final outcome.

These five steps can be logically thought of as running in sequence – each step starts only after the previous step is completed – although in practice they can be interleaved as long as the final result is not affected.

In many situations, the input data might have already been distributed ("sharded") among many different servers, in which case step 1 could sometimes be greatly simplified by assigning Map servers that would process the locally present input data. Similarly, step 3 could sometimes be sped up by assigning Reduce processors that are as close as possible to the Map-generated data they need to process.

The Map and Reduce functions of MapReduce are both defined with respect to data structured in (key, value) pairs. Map takes one pair of data with a type in one data domain, and returns a list of pairs in a different domain:

Map(k1,v1) → list(k2,v2)

The Map function is applied in parallel to every pair (keyed by k1) in the input dataset. This produces a list of pairs (keyed by k2) for each call. After that, the MapReduce framework collects all pairs with the same key (k2) from all lists and groups them together, creating one group for each key.

The Reduce function is then applied in parallel to each group, which in turn produces a collection of values in the same domain:

Reduce(k2, list (v2)) → list((k3, v3))^[15]

Each Reduce call typically produces either one key value pair or an empty return, though one call is allowed to return more than one key value pair. The returns of all calls are collected as the desired result list.

Thus the MapReduce framework transforms a list of (key, value) pairs into another list of (key, value) pairs.^[16] This behavior is different from the typical functional programming map and reduce combination, which accepts a list of arbitrary values and returns one single value that combines all the values returned by map.

It is necessary but not sufficient to have implementations of the map and reduce abstractions in order to implement MapReduce. Distributed implementations of MapReduce require a means of connecting the processes performing the Map and Reduce phases. This may be a distributed file system. Other options are possible, such as direct streaming from mappers to reducers, or for the mapping processors to serve up their results to reducers that query them.

functionmap(String name, String document): // name: document name// document: document contentsfor each word w in document: emit (w, 1) function reduce (String word, Iterator partialCounts): // word: 単語// partialCounts: 集計された部分カウントのリスト 合計 = 0 partialCountsの各pcについて: 合計 += pc （単語、合計）を放出する

SELECT age , AVG ( contacts ) FROM social.person GROUP BY age ORDER BY age​​

関数Map は、1 ～ 1100 の 整数K1 を入力として受け取ります。K1 バッチ内の各 social.person レコードに対して、 Y を人物の年齢とし 、Nを人物の連絡先の数とし、 出力レコード(Y,(N,1)) を 1 つ生成し、繰り返します。関数終了関数Reduceは入力として年齢 (年) Y を 受け取り、各入力レコード (Y,(N,C))に対して以下の処理を実行します。 SにN*Cの合計を蓄積するC new にC の 合計 を蓄積します。Aを S/C newとし、1 つの出力レコード(Y,(A,C new )) を生成します。関数終了

-- map output #1: age, quantity of contacts 10, 9 10, 9 10, 9

-- map output #2: age, quantity of contacts 10, 9 10, 9

-- map output #3: age, quantity of contacts 10, 10

-- reduce step #1: age, average of contacts 10, 9

Software framework architecture adheres to open-closed principle where code is effectively divided into unmodifiable frozen spots and extensiblehot spots. The frozen spot of the MapReduce framework is a large distributed sort. The hot spots, which the application defines, are:

• an input reader
• a Map function
• a partition function
• a compare function
• a Reduce function
• an output writer

Properties of monoids are the basis for ensuring the validity of MapReduce operations.^[18][19]

In the Algebird package^[20] a Scala implementation of Map/Reduce explicitly requires a monoid class type .^[21]

The operations of MapReduce deal with two types: the type A of input data being mapped, and the type B of output data being reduced.

The Map operation takes individual values of type A and produces, for each a:A a value b:B; The Reduce operation requires a binary operation • defined on values of type B; it consists of folding all available b:B to a single value.

From a basic requirements point of view, any MapReduce operation must involve the ability to arbitrarily regroup data being reduced. Such a requirement amounts to two properties of the operation •:

• associativity: (x • y) • z = x • (y • z)
• existence of neutral element e such that e • x = x • e = x for every x:B.

The second property guarantees that, when parallelized over multiple nodes, the nodes that don't have any data to process would have no impact on the result.

These two properties amount to having a monoid (B, •, e) on values of type B with operation • and with neutral element e.

There's no requirements on the values of type A; an arbitrary function A→B can be used for the Map operation. This means that we have a catamorphismA*→ (B, •, e). Here A* denotes a Kleene star, also known as the type of lists over A.

The Shuffle operation per se is not related to the essence of MapReduce; it's needed to distribute calculations over the cloud.

It follows from the above that not every binary Reduce operation will work in MapReduce. Here are the counter-examples:

• building a tree from subtrees: this operation is not associative, and the result will depend on grouping;
• direct calculation of averages: avg is also not associative (and it has no neutral element); to calculate an average, one needs to calculate moments.

MapReduce programs are not guaranteed to be fast. The main benefit of this programming model is to exploit the optimized shuffle operation of the platform, and only having to write the Map and Reduce parts of the program. In practice, the author of a MapReduce program however has to take the shuffle step into consideration; in particular the partition function and the amount of data written by the Map function can have a large impact on the performance and scalability. Additional modules such as the Combiner function can help to reduce the amount of data written to disk, and transmitted over the network. MapReduce applications can achieve sub-linear speedups under specific circumstances.^[22]

When designing a MapReduce algorithm, the author needs to choose a good tradeoff^[11] between the computation and the communication costs. Communication cost often dominates the computation cost,^[11][22] and many MapReduce implementations are designed to write all communication to distributed storage for crash recovery.

In tuning performance of MapReduce, the complexity of mapping, shuffle, sorting (grouping by the key), and reducing has to be taken into account. The amount of data produced by the mappers is a key parameter that shifts the bulk of the computation cost between mapping and reducing. Reducing includes sorting (grouping of the keys) which has nonlinear complexity. Hence, small partition sizes reduce sorting time, but there is a trade-off because having a large number of reducers may be impractical. The influence of split unit size is marginal (unless chosen particularly badly, say <1MB). The gains from some mappers reading load from local disks, on average, is minor.^[23]

For processes that complete quickly, and where the data fits into main memory of a single machine or a small cluster, using a MapReduce framework usually is not effective. Since these frameworks are designed to recover from the loss of whole nodes during the computation, they write interim results to distributed storage. This crash recovery is expensive, and only pays off when the computation involves many computers and a long runtime of the computation. A task that completes in seconds can just be restarted in the case of an error, and the likelihood of at least one machine failing grows quickly with the cluster size. On such problems, implementations keeping all data in memory and simply restarting a computation on node failures or —when the data is small enough— non-distributed solutions will often be faster than a MapReduce system.

MapReduce achieves reliability by parceling out a number of operations on the set of data to each node in the network. Each node is expected to report back periodically with completed work and status updates. If a node falls silent for longer than that interval, the master node (similar to the master server in the Google File System) records the node as dead and sends out the node's assigned work to other nodes. Individual operations use atomic operations for naming file outputs as a check to ensure that there are not parallel conflicting threads running. When files are renamed, it is possible to also copy them to another name in addition to the name of the task (allowing for side-effects).

The reduce operations operate much the same way. Because of their inferior properties with regard to parallel operations, the master node attempts to schedule reduce operations on the same node, or in the same rack as the node holding the data being operated on. This property is desirable as it conserves bandwidth across the backbone network of the datacenter.

Implementations are not necessarily highly reliable. For example, in older versions of Hadoop the NameNode was a single point of failure for the distributed filesystem. Later versions of Hadoop have high availability with an active/passive failover for the "NameNode."

MapReduce is useful in a wide range of applications, including distributed pattern-based searching, distributed sorting, web link-graph reversal, singular value decomposition,^[24] web access log stats, inverted index construction, document clustering, machine learning,^[25] and statistical machine translation. Moreover, the MapReduce model has been adapted to several computing environments like multi-core and many-core systems,^[26][27][28] desktop grids,^[29] multi-cluster,^[30] volunteer computing environments,^[31] dynamic cloud environments,^[32] mobile environments,^[33] and high-performance computing environments.^[34]

At Google, MapReduce was used to completely regenerate Google's index of the World Wide Web. It replaced the old ad hoc programs that updated the index and ran the various analyses.^[35] Development at Google has since moved on to technologies such as Percolator, FlumeJava^[36] and MillWheel that offer streaming operation and updates instead of batch processing, to allow integrating "live" search results without rebuilding the complete index.^[37]

MapReduce's stable inputs and outputs are usually stored in a distributed file system. The transient data are usually stored on local disk and fetched remotely by the reducers.

• Bird–Meertens formalism
• Parallelization contract