Какая компания создала технологию mapreduce

MapReduce is a programming model and an associated implementation for processing and generating big data sets with a parallel, distributed algorithm on a cluster.^[1][2][3]

A MapReduce program is composed of a map procedure, which performs filtering and sorting (such as sorting students by first name into queues, one queue for each name), and a reduce method, which performs a summary operation (such as counting the number of students in each queue, yielding name frequencies). The «MapReduce System» (also called «infrastructure» or «framework») orchestrates the processing by marshalling the distributed servers, running the various tasks in parallel, managing all communications and data transfers between the various parts of the system, and providing for redundancy and fault tolerance.

The model is a specialization of the split-apply-combine strategy for data analysis.^[4]
It is inspired by the map and reduce functions commonly used in functional programming,^[5] although their purpose in the MapReduce framework is not the same as in their original forms.^[6] The key contributions of the MapReduce framework are not the actual map and reduce functions (which, for example, resemble the 1995 Message Passing Interface standard’s^[7] reduce^[8] and scatter^[9] operations), but the scalability and fault-tolerance achieved for a variety of applications by optimizing the execution engine^{[citation needed]}. As such, a single-threaded implementation of MapReduce is usually not faster than a traditional (non-MapReduce) implementation; any gains are usually only seen with multi-threaded implementations on multi-processor hardware.^[10] The use of this model is beneficial only when the optimized distributed shuffle operation (which reduces network communication cost) and fault tolerance features of the MapReduce framework come into play. Optimizing the communication cost is essential to a good MapReduce algorithm.^[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 been genericized. By 2014, Google was no longer using MapReduce as their 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]

Overview[edit]

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.

Logical view[edit]

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.

Examples[edit]

The canonical MapReduce example counts the appearance of each word in a set of documents:^[17]

function map(String name, String document):
    // name: document name
    // document: document contents
    for each word w in document:
        emit (w, 1)

function reduce(String word, Iterator partialCounts):
    // word: a word
    // partialCounts: a list of aggregated partial counts
    sum = 0
    for each pc in partialCounts:
        sum += pc
    emit (word, sum)

Here, each document is split into words, and each word is counted by the map function, using the word as the result key. The framework puts together all the pairs with the same key and feeds them to the same call to reduce. Thus, this function just needs to sum all of its input values to find the total appearances of that word.

As another example, imagine that for a database of 1.1 billion people, one would like to compute the average number of social contacts a person has according to age. In SQL, such a query could be expressed as:

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

Using MapReduce, the K1 key values could be the integers 1 through 1100, each representing a batch of 1 million records, the K2 key value could be a person’s age in years, and this computation could be achieved using the following functions:

function Map is
    input: integer K1 between 1 and 1100, representing a batch of 1 million social.person records
    for each social.person record in the K1 batch do
        let Y be the person's age
        let N be the number of contacts the person has
        produce one output record (Y,(N,1))
    repeat
end function

function Reduce is
    input: age (in years) Y
    for each input record (Y,(N,C)) do
        Accumulate in S the sum of N*C
        Accumulate in Cnew the sum of C
    repeat
    let A be S/Cnew
    produce one output record (Y,(A,Cnew))
end function

Note that in the Reduce function, C is the count of people having in total N contacts, so in the Map function it is natural to write C=1, since every output pair is referring to the contacts of one single person.

The MapReduce system would line up the 1100 Map processors, and would provide each with its corresponding 1 million input records. The Map step would produce 1.1 billion (Y,(N,1)) records, with Y values ranging between, say, 8 and 103. The MapReduce System would then line up the 96 Reduce processors by performing shuffling operation of the key/value pairs due to the fact that we need average per age, and provide each with its millions of corresponding input records. The Reduce step would result in the much reduced set of only 96 output records (Y,A), which would be put in the final result file, sorted by Y.

The count info in the record is important if the processing is reduced more than one time. If we did not add the count of the records, the computed average would be wrong, for example:

-- 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

If we reduce files #1 and #2, we will have a new file with an average of 9 contacts for a 10-year-old person ((9+9+9+9+9)/5):

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

If we reduce it with file #3, we lose the count of how many records we’ve already seen, so we end up with an average of 9.5 contacts for a 10-year-old person ((9+10)/2), which is wrong. The correct answer is 9.166 = 55 / 6 = (9×3+9×2+10×1)/(3+2+1).

Dataflow[edit]

Software framework architecture adheres to open-closed principle where code is effectively divided into unmodifiable frozen spots and extensible hot 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

Input reader[edit]

The input reader divides the input into appropriate size ‘splits’ (in practice, typically, 64 MB to 128 MB) and the framework assigns one split to each Map function. The input reader reads data from stable storage (typically, a distributed file system) and generates key/value pairs.

A common example will read a directory full of text files and return each line as a record.

Map function[edit]

The Map function takes a series of key/value pairs, processes each, and generates zero or more output key/value pairs. The input and output types of the map can be (and often are) different from each other.

If the application is doing a word count, the map function would break the line into words and output a key/value pair for each word. Each output pair would contain the word as the key and the number of instances of that word in the line as the value.

Partition function[edit]

Each Map function output is allocated to a particular reducer by the application’s partition function for sharding purposes. The partition function is given the key and the number of reducers and returns the index of the desired reducer.

A typical default is to hash the key and use the hash value modulo the number of reducers. It is important to pick a partition function that gives an approximately uniform distribution of data per shard for load-balancing purposes, otherwise the MapReduce operation can be held up waiting for slow reducers to finish
(i.e. the reducers assigned the larger shares of the non-uniformly partitioned data).

Between the map and reduce stages, the data are shuffled (parallel-sorted / exchanged between nodes) in order to move the data from the map node that produced them to the shard in which they will be reduced. The shuffle can sometimes take longer than the computation time depending on network bandwidth, CPU speeds, data produced and time taken by map and reduce computations.

Comparison function[edit]

The input for each Reduce is pulled from the machine where the Map ran and sorted using the application’s comparison function.

Reduce function[edit]

The framework calls the application’s Reduce function once for each unique key in the sorted order. The Reduce can iterate through the values that are associated with that key and produce zero or more outputs.

In the word count example, the Reduce function takes the input values, sums them and generates a single output of the word and the final sum.

Output writer[edit]

The Output Writer writes the output of the Reduce to the stable storage.

Theoretical background[edit]

Properties of Monoid are the basis for ensuring the validity of Map/Reduce operations.^[18][19]

In Algebird package^[20] a Scala implementation of Map/Reduce explicitly requires 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.

Map operation takes individual values of type A and produces, for each a:A a value b:B; 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 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 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 Map operation. This means that we have a Catamorphism A* → (B, •, e). Here A* denotes a Kleene star, also known as the type of lists over A.

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.

Performance considerations[edit]

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.

Distribution and reliability[edit]

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.»

Uses[edit]

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.

Criticism[edit]

Lack of novelty[edit]

David DeWitt and Michael Stonebraker, computer scientists specializing in parallel databases and shared-nothing architectures, have been critical of the breadth of problems that MapReduce can be used for.^[38] They called its interface too low-level and questioned whether it really represents the paradigm shift its proponents have claimed it is.^[39] They challenged the MapReduce proponents’ claims of novelty, citing Teradata as an example of prior art that has existed for over two decades. They also compared MapReduce programmers to CODASYL programmers, noting both are «writing in a low-level language performing low-level record manipulation.»^[39] MapReduce’s use of input files and lack of schema support prevents the performance improvements enabled by common database system features such as B-trees and hash partitioning, though projects such as Pig (or PigLatin), Sawzall, Apache Hive,^[40] HBase^[41] and Bigtable^[41][42] are addressing some of these problems.

Greg Jorgensen wrote an article rejecting these views.^[43] Jorgensen asserts that DeWitt and Stonebraker’s entire analysis is groundless as MapReduce was never designed nor intended to be used as a database.

DeWitt and Stonebraker have subsequently published a detailed benchmark study in 2009 comparing performance of Hadoop’s MapReduce and RDBMS approaches on several specific problems.^[44] They concluded that relational databases offer real advantages for many kinds of data use, especially on complex processing or where the data is used across an enterprise, but that MapReduce may be easier for users to adopt for simple or one-time processing tasks.

The MapReduce programming paradigm was also described in Danny Hillis’s 1985 thesis ^[45] intended for use on the Connection Machine, where it was called «xapping/reduction»^[46] and relied upon that machine’s special hardware to accelerate both map and reduce. The dialect ultimately used for the Connection Machine, the 1986 StarLisp, had parallel *map and reduce!!,^[47] which in turn was based on the 1984 Common Lisp, which had non-parallel map and reduce built in.^[48] The tree-like approach that the Connection Machine’s hypercube architecture uses to execute reduce in time^[49] is effectively the same as the approach referred to within the Google paper as prior work.^[3]: 11

In 2010 Google was granted what is described as a patent on MapReduce. The patent, filed in 2004, may cover use of MapReduce by open source software such as Hadoop, CouchDB, and others. In Ars Technica, an editor acknowledged Google’s role in popularizing the MapReduce concept, but questioned whether the patent was valid or novel.^[50][51] In 2013, as part of its «Open Patent Non-Assertion (OPN) Pledge», Google pledged to only use the patent defensively.^[52][53] The patent is expected to expire on 23 December 2026.^[54]

Restricted programming framework[edit]

MapReduce tasks must be written as acyclic dataflow programs, i.e. a stateless mapper followed by a stateless reducer, that are executed by a batch job scheduler. This paradigm makes repeated querying of datasets difficult and imposes limitations that are felt in fields such as graph processing^[55] where iterative algorithms that revisit a single working set multiple times are the norm, as well as, in the presence of disk-based data with high latency, even the field of machine learning where multiple passes through the data are required even though algorithms can tolerate serial access to the data each pass.^[56]

See also[edit]

• Bird–Meertens formalism

Implementations of MapReduce[edit]

• Apache CouchDB
• Apache Hadoop
• Infinispan
• Riak

References[edit]

Логистика — замечательная штука. Стоит заглянуть в хранилище какого-нибудь гипермаркета — и становится ясно: без тщательного упорядочивания товаров в нём и речи не может идти о какой-либо эффективной торговле. Между тем гигантские торговые дома преспокойно открывают свои двери десяткам тысяч потребителей и легко разыскивают в недрах своих бездонных складов требуемые товары.

С информацией дело обстоит примерно так же. Сохранить её — только одна из задач. Информация требует обработки. Каким бы эффективным ни было хранилище данных, оно не поможет её обработать.

В компании Google с этой проблемой знакомы не понаслышке. Её трудолюбивые боты круглосуточно собирают контент всё более разрастающегося интернета и передают его в кластер Google, где правит бал распределённая файловая система GFS. Она распределяет петабайты данных по сотням тысяч чанк-серверов не хуже матёрого специалиста по логистике, по ходу дела обеспечивая им надежное хранение, которому не вредят ни возможные сбои, ни отказы оборудования.

Но поисковая система — это не столько склад контента, сколько система быстрого нахождения его частей по ключевым словам, вводимым пользователем в строку поиска.

Как узнать, на каких веб-страницах есть эти ключевые слова? Какой ресурс содержит всего одно ключевое слово, а какой — сразу несколько? Тут без хорошей индексной базы не обойтись.

Сформировать такую базу на петабайтном массиве данных — задачка нетривиальная. Фактически требуется поочередно «пройтись» по каждому документу и определить, содержит ли он требуемое слово. Как ни крути, процесс линейный.

Но только не для компании, обладающей кластером из полумиллиона компьютеров.

С ним легко можно применять известную парадигму информатики «разделяй и властвуй», поделив информационный массив на части и отдав каждую из них индексировать отдельному серверу. Ну а после выполнения индексации по частям остается собрать найденное решение воедино.

Именно для такой работы Google и была разработана технология MapReduce — средство эффективного распараллеливания задач, в обычных условиях решаемых линейно.

Вдохновение из прошлого

Помните, в детстве на уроках родной речи нам давали задание на внимательность? Найти, например, на странице из «Золотой рыбки» все буквы «О» и подсчитать их количество. Школьники скрупулёзно, строка за строкой, высматривали «О», подчеркивали их, а потом суммировали подчёркнутое.

Теперь представьте, насколько быстрее пошло бы дело, если бы мы раздали по строке каждому ученику из класса, который, подсчитав количество букв, сообщил бы его заранее назначенному «главному по буквам». Если не вдаваться в подробности, технология MapReduce работает именно таким образом.

Конечно, разрабатывая MapReduce, сотрудники Google Джеффри Дин (Jeffrey Dean) и Санджай Гемават (Sanjay Ghemawat) вдохновлялись не школьными заданиями. Они имели другой замечательный источник энтузиазма. Ещё в конце пятидесятых годов прошлого столетия для обработки данных, представленных линейными списками, был разработан язык программирования Lisp — предок целого семейства языков, использующих парадигму функционального программирования.

Lisp и другие функциональные языки поддерживают интересную программную модель, именуемую Map/Reduce. Из её названия следует, что работают в ней две процедуры — map, применяющая нужную функцию к каждому элементу списка, и reduce, объединяющая результаты работы map.

Замечательным свойством модели Map/Reduce является всеядность. С её помощью, например, удобно организовать счётчик появления искомых слов в большом файле (построение Term-вектора) или счётчик частоты обращений к заданному адресу, вычислить объём всех веб-страниц со всех URL конкретного хост-узла или же создать список всех адресов, содержащих необходимые данные.

Чтобы выполнить индексацию сохраняемого в кластере Google веб-контента, разработчики MapReduce приспособили общую модель функциональных языков программирования к своим целям.

Они предположили, что вся обрабатываемая MapReduce информация может быть представлена в виде списка пар, каждая из которых состоит из ключа и значения. В массиве веб-контента ключом, конечно же, является искомое слово, а его значением — число 1, подтверждающее присутствие этого слова.

В самом простом варианте программной модели Google MapReduce процедура map получает на вход список слов, содержащихся в обрабатываемых документах. Она преобразует каждый элемент в пару, одним элементом которой является слово, а другим — число 1.

Затем за список берётся процедура reduce, которая группирует элементы списка с одинаковыми ключами (то есть словами) и суммирует единички. В результате на выходе оказывается список всех ключевых слов с количеством их упоминаний в тех или иных документах. А это и есть индексная база поисковой системы.

Как видите, ничего нового создатели MapReduce не изобрели. Их настоящая заслуга в другом. На самом деле MapReduce — это не только программная модель, используя которую можно решать задачи сортировки и группировки данных. Это — целая архитектура, обеспечивающая:

• автоматическое распараллеливание данных из огромного массива по множеству узлов обработки, выполняющих процедуры Map/Reduce;
• эффективную балансировку загрузки этих вычислительных узлов, не дающую им простаивать или быть перегруженными сверх меры;
• технологию отказоустойчивой работы, предусматривающую тот факт, что при выполнении общего задания часть узлов обработки может выйти из строя или по какой-либо другой причине перестать обрабатывать данные.

Таким образом, Google MapReduce, с одной стороны, предоставляет пользователю процедуры обработки его данных, а с другой — делает для него прозрачным процесс распараллеливания этой обработки на могучем кластере Google.

Как же им это удалось?

Параллельные вычислители, GFS и всё тот же кластер

Наиболее светлой мыслью при проектировании MapReduce была идея разместить модули, реализующие процедуры map и reduce, на тех самых чанк-серверах — основе файловой системы GFS. Такой подход приближает хранящиеся в GPS модули к функциям их обработки. Экономия сетевого трафика налицо.

Дальше — больше. Как и GFS, технология MapReduce построена по принципу «главный — подчиненные». «Голова» MapReduce — процедура Master — управляет множеством разбросанных по чанк-серверам «работников» (workers), часть из которых отвечает за функцию map (их зовут mappers, или «мэпперы»), а остальные, соответственно, за reduce (reducers — «редьюсеры»).

На вход MapReduce поступает требующий обработки массив, «разрезанный» на M (по числу мэпперов) частей размером от 16 до 64 мегабайт (стоит напомнить, что именно такой размер имеет чанк в файловой системе GFS). Получив адреса M частей массива, Master MapReduce формирует частные задания для M функций мэпперов и раздает каждой из них адрес чанка, который надлежит подвергнуть процедуре map. Поскольку мэпперы работают параллельно и независимо друг от друга, требуется в M раз меньше времени, чем при линейной обработке.

В результате появляется новый, разделённый на части массив промежуточных данных, содержащих неупорядоченные списки пар ключ — значение. В идеале количество частей этого промежуточного массива должно быть равно R, то есть совпадать с количеством «работников», отвечающих за операцию reduce. Однако на практике массив пар, содержащих один и тот же ключ, может быть значительно больше (например, если ключ — это одно из самых популярных слов в поисковых запросах).

Чтобы сократить его размер, MapReduce использует процедуру предварительного агрегирования данных, присваивая таким популярным парам новое промежуточное значение. Эта процедура именуется combine и по своей сути очень похожа на reduce. Необходимо заметить, что combine можно использовать лишь в тех случаях, когда функция, которую используют на стадии reduce для объединения данных, обладает свойствами коммутативности и ассоциативности.

Агрегированный до требуемого размера массив промежуточных данных может поступать на R «работников», выполняющих reduce. Стоит напомнить, что reduce в простейшем виде работает со всеми значениями одного ключа — например, с количеством упоминаний какого-либо слова. Это значит, что на каждого «работника» желательно подать пары с одинаковым ключом. Проблема заключается в том, что они разбросаны по разным частям списка, сформированного мэпперами. Как же быть?

Последним этапом перед выполнением процедуры reduce является процедура распределения (partitioning), в результате которой пары с одинаковым ключом попадают на одних и тех же «работников». Да, процесс требует времени и значительного сетевого трафика, но всё это компенсируется скоростью работы на следующем этапе.

Каждый редьюсер в конечном итоге создает файл, где хранятся отсортированный (например, по алфавиту) список ключей, за которые он отвечал, и результат обработки значений этих ключей (например, их сумма).

R «работников» создают R результирующих файлов, о чём и докладывают мастеру MapReduce. Получив подтверждение от всех «работников», он считает задание выполненным и передает адреса результирующих файлов клиентскому приложению.

Так в недрах кластера Google и рождается индексная база интернета, строятся графы популярности URL и выполняются прочие необходимые для работы поисковой системы операции.

И не только для неё. Разработчики приложений Google, в которых требуется обработка больших массивов данных, быстро оценили достоинства параллельного вычислителя MapReduce и написали свои варианты процедур map и reduce. Именно так пополняется словарная база переводчика Google Translate, так формирует данные об орфографии десятков языков спеллчекер Google Docs и индексирует голосовые морфемы Google Voice. Применений MapReduce было найдено множество.

Как и в случае с файловой системой GFS, Google пояснила только идеологию MapReduce, не раскрывая конкретных реализаций её алгоритмов. Однако и этого хватило, чтобы сообщество open source в рамках проекта Hadoop реализовало планировщик распределения задач по узлам вычислительного кластера Hadoop YARN и фреймворк Hadoop MapReduce, позволяющий создавать собственные мэпперы и редьюсеры.

Вскоре MapReduce начала обрабатывать «большие данные» в других поисковых системах, социальных сетях и интернет-магазинах.

Между тем с момента разработки этой технологии интернет серьёзно изменился. Как в плане контента, где наряду с текстовыми данными всё большую роль стали играть изображения, видео, звук, картографическая информация, так и в плане социальном. Нынешний интернет-житель в качестве стартовой страницы своего браузера всё реже ставит поисковую систему и всё чаще — социальную сеть. Контент нынешнего интернета более динамичный: он меняется буквально ежесекундно.

Технология же MapReduce, ориентированная на пакетную обработку данных (мастер раздал задания, подождал, собрал результаты) не могла эффективно учитывать эти постоянные изменения.

Именно поэтому в июле 2010 года Google представила свою новую систему распределенного индексирования интернет-контента Caffeine. В ней трудится обновленный вариант технологии MapReduce, ориентированный на инкрементное формирование распределённой индексной базы.

Теперь индекс Google обновляется не единовременно, но редко, а понемногу, но часто. И это позволяет проиндексировать такие мимолетные, казалось бы, ресурсы, как посты социальных сетей или многочисленные навигационные запросы к сервису Google Maps.

Используя свои лучшие наработки и тщательно исследуя тенденции нашей с вами работы в интернете, Google старается делать упреждающие шаги в развитии своего сердца — поисковой системы.

Большинство отраслевых аналитиков сходятся во мнении, что с периодом примерно в шесть лет суммарное количество данных, накапливаемых в компьютерных базах, увеличивается на три порядка, а это существенно выше прироста производительности компьютеров. С определенной натяжкой можно утверждать, что производительность процессоров увеличивается за те же шесть лет на один порядок – терабайтные диски позволяют создать петабайтное хранилище без особой сложности, но сдерживающим фактором становится скорость обработки.

Cкорость доступа к данным можно было бы поднять в сотни и более раз, если перейти от последовательных схем к параллельным, и предпосылки к таким решениям уже существуют: кластерные конфигурации, многоядерные процессоры (multicore и manycore). По некоторым прогнозам, именно базы данных окажутся «убийственным приложением» для многоядерных процессоров и расчистят дорогу новому поколению процессоров. В отличие от многих других приложений, работа с данными распараллеливается более естественным образом. Разделить базу на фрагменты, передать их обработку на узлы кластера или в ядра многоядерного процессора, а затем каким-то образом собрать полученные промежуточные результаты и сформулировать решение проще, чем решить традиционную для кластеров задачу распараллеливания сложной модели.

База данных без общих ресурсов

Если база данных построена на общем информационном ресурсе, то разделить ее на части для параллельной обработки невозможно, следовательно, предметом распараллеливания должна быть не классическая реляционная база, а база, изначально рассчитанная на архитектуру без разделения ресурсов (Shared Nothing, SN). До сих пор у специалистов по СУБД архитектура SN оставалась на периферии внимания, впрочем, иногда ее противопоставляют централизованным решениям, сравнивают с решениями на основе реляционных СУБД и серверов приложений, но не рассматривают в качестве реального конкурента им. Чаще всего представление о SN ограничивают различного рода Web-приложениями. Это покажется особенно удивительным, если вспомнить, что первым идею базы данных с архитектурой SN выдвинул сам Майкл Стоунбрейкер, один из первопроходцев реляционных СУБД, причем сделал это задолго до появления WWW. В 1986 году в статье The Case for Shared Nothing он выделил три категории систем, способных хранить данные: с разделяемой памятью (Shared Memory, SM), с разделяемыми дисками (Shared Disk, SD) и не разделяющими ничего (SN). Первым двум соответствовало множество различных систем, а архитектура SN была представлена только сетевыми по своей природе компьютерами Tandem (сегодня их наследниками являются HP NonStop Himalaya S-Series). И все же Стоунбрейкеру удалось показать, что гипотетически архитектура SN наиболее предпочтительна, в том числе и с экономической точки зрения. Но двадцать лет назад сетевые архитектуры не были достаточно развиты, а решения Tandem, несмотря на всю их изящность, оставались проприетарными и в силу этого слишком дорогими. Нет ничего удивительного, что эволюция серверов пошла иным путем. Но сегодня к SN возвращаются уже на другом витке развития технологий, и в данном случае приоритет возрождения принадлежит Google, архитектура SN воплощена в конструкции, называемой MapReduce.

Приложения, построенные на основе MapReduce, в принципе можно рассматривать и как базы данных, но не в том смысле, что в их основе обязательно лежит реляционная модель с доступом к данным с помощью языка SQL, а как опирающиеся на альтернативную модель и рассчитанные на очень большие хранилища данных.

Три архитектуры

При создании аппаратных систем, поддерживающих базы данных, возможен выбор между тремя архитектурными вариантами: share-everything, share-disk и share-nothing (рис. 1).

Рис. 1. Архитектуры хранилищ данных

Наиболее массовыми являются решения share-everything, где разделяются все ресурсы. Они предполагают использование универсальных серверов с множеством симметричных процессоров SMP (Symmetric Multi-Processing) в сочетании с реляционными, реже объектными или постреляционными СУБД. В этих решениях воплощены основные идеологические принципы, на которых строились компьютеры для корпоративных приложений, – с конца 60-х годов качество компьютеров оценивалось по их способности к распределению ресурсов. Реляционные базы создавались в расчете на транзакционную обработку данных (OLTP) на мощных SMP-серверах, обеспечивающих полное распределение всех ресурсов. Даже когда в девяностых годах стали постепенно осознавать достоинства систем с массовым параллелизмом (MPP), с точки зрения СУБД преимущество сохранялось за SMP, главным образом из-за преемственности и совместимости программного обеспечения. Но со временем с большей остротой стали проявляться два основных недостатка shared-everything. Каждый процессор должен быть осведомлен о деятельности остальных – часть мощности уходит на поддержку «коммунальной жизни», что в конечном итоге задает предел числа процессоров в одной системе, поскольку издержки растут экспоненциально. Пользование общим дисковым пространством ведет к перегрузке канала, связывающего процессоры с памятью, и серверы этого типа имеют врожденный предел роста.

В качестве одной из альтернатив SMP-серверам можно рассматривать кластерные архитектуры, в которых распределяемым является только дисковое пространство, системы этого класса поддерживаются, например, в Oracle RAC (Real application Cluster). Кластер с разделяемыми дисками, на котором работает хранилище данных, собирается из нескольких серверов и через SAN подключается к распределяемым системам хранения данных. Наиболее полно RAC реализован в продукте HP Oracle Database Machine. Эта машина баз данных состоит из программных компонентов от Oracle, серверов стандартной архитектуры от HP и структурно представляет собой две grid-системы – на одной (8 серверов Oracle Database Server на HP Proliant DL360 G5) работает СУБД, а вторая (14 серверов Oracle Exadata Storage на HP ProLiant DL180 G5) предназначена для хранения данных. Эти специализированные серверы отличаются от обычных Linux-серверов тем, что в них встроено программное обеспечение, позволяющее часть функций по обработке запросов выполнять в параллельном режиме, то есть некоторая часть работы с базой выполняется локально, что естественным образом ускоряет работу. Вместе две системы объединены через InfiniBand. Избранная архитектура и высокоскоростное межсоединение избавляют HP Oracle Database Machine от проблем, связанных с «бутылочным горлом».

Третье решение – использование стандартных серверов по принципу «никаких общих ресурсов» . Подобные архитектуры отлично зарекомендовали себя применительно к высокопроизводительным вычислениям, но их внедрение в других областях ограничивается сложностями, связанными с распараллеливанием приложений. Идея баз данных без общих ресурсов (shared-nothing database) в течение длительного срока жила в продуктах компании Teradata, выпускающей MPP-системы с собственной коммутационной инфраструктурой BYNET. До последнего времени единственной альтернативой изделиям от Teradata были специализированные серверы Netezza Performance Server (NPS). Верхний уровень этих серверов – хост на базе SMP-сервера, работающий под управлением Linux, а нижний уровень – сотни лезвий Snippet Processing Unit (SPU), образующих кластер. Задача хоста состоит в преобразовании пользовательских запросов в лоскутные SPU и возврате результатов пользователям. Каждый из SPU отвечает за выполнение отдельной части запроса. Эти системы ориентированы в основном на бизнес-аналитику.

Недавно к этой небольшой группе разработчиков присоединилась компания Greenplum, которая начала с нуля, но подошла к делу радикально, собрав команду высококвалифицированных специалистов и написав совершенно новую СУБД. Отличие альтернативной архитектуры в том, что каждый отдельный компонент этой СУБД играет роль законченной, самодостаточной мини-СУБД, которая сама владеет и оперирует отдельной порцией доверенных ей данных. Эта мини-СУБД автоматически распределяет данные и нагрузку по обработке запросов по доступным ей аппаратными ресурсам. Для своего старта компания Greenplum избрала удачный с точки зрения технической конъюнктуры момент – она решила стать первым поставщиком «убийственного приложения» многоядерных процессоров. Во главу угла ее стратегии поставлена идея сборки систем, аналогичных по своей функциональности продуктам компаний Teradata и Netezza, но отличающихся тем, что в них используется широко распространенное аппаратное обеспечение. Можно сказать, что Greenplum решила повторить Google, но применительно к базам данных. Иными словами, разработчики Greenplum решили обеспечить доступ к данным не только средствами поиска, но и через SQL. Поскольку Greenplum исключительно программная компания, основным моментом заимствования стала программная конструкция MapReduce, разработанная в Google.

Технологическое начинание Greenplum заслужило высокую оценку в экспертном сообществе – в отчете Gartner «Магический квадрант СУБД для хранилищ данных» от 24 января 2008 года лишь эта компания находится в части квадранта visionaries (провидцы), что свидетельствует о несомненной перспективности, но еще недостаточной массовости предлагаемых технологий или продуктов. В отчете о Greenplum подчеркивается как достоинство монопродуктовая ориентация – компания поставляет исключительно системы управления базами данных для хранилищ. Ее продукты ориентированы на системы с массовым параллелизмом, в них используется модернизированная версия PostgreSQL, объектно-реляционной СУБД с открытыми кодами, созданной в конце 90-х годов как логическое продолжение СУБД Ingress. Продукция Greenplum распространяется в комплекте со специализированными программируемыми устройствами (appliance). Поставщиком оборудования является Sun Microsystems – две компании выпускают Sun Data Warehouse Appliance, состоящие из специализированных серверов Sun Fire X4540 и Greenplum Dtabase.

Введение в MapReduce

Программная конструкция MapReduce пошла от Google, известной не только своими технологическими достижениями, но и способностью их скрывать. Именно в таком духе выдержана статья Джефри Дина и Санжая Чемавата «MapReduce: упрощенная обработка данных на больших кластерах» (MapReduce: Simplied Data Processing on Large Clusters), недосказанности которой компенсирует работа Ральфа Ламмеля из Исследовательского центра Microsoft в Редмонде «Модель программирования MapReduce компании Google» (Google’s MapReduce Programming Model). Во введении Ламмель заявляет, что он проанализировал статью Дина и Чемавата, которую он неоднократно и с почтением называет seminal (судьбоносной), используя метод, который называется «обратной инженерией» (reverse engineering). То есть, не выходя за рамки ограничений на авторские права, он постарался раскрыть смысл того, что же в ней на самом деле представлено. Ламмель увязывает MapReduce с функциональным программированием, языками Лисп и Haskell, что естественным образом вводит читателя в контекст.

Но работа Ламмеля ориентирована исключительно на программистов-разработчиков, поэтому, прежде чем изложить в адаптированной форме программную концепцию MapReduce, остановимся на том, что у Ламмеля не сказано, но что критически важно для понимания предпосылок. Дело в том, что MapReduce – это не просто программный каркас (framework), как ее предшественники в Лисп, а скорее инфраструктурное решение, способное эффективно использовать сегодня кластерные, а в будущем многоядерные архитектуры, своей перспективностью она и привлекает к себе повышенное внимание.

Почему же значимость MapReduce выходит за пределы тех приложений, для которых этот программный каркас используется в Google, и привлекает к себе всеобщее внимание? Те огромные кластеры, на которых она применяется, и поисковые задачи на самом деле всего лишь частный случай. По мнению одного из выдающихся умов современности, профессора Калифорнийского университета
в Беркли Дейва Паттерсона, MapReduce станет одним из основных инструментов программирования в будущем (см. презентацию The Parallel Revolution Has Started), когда жизнь будет подчиняться не только классическому, но и модифицированному закону Мура, утверждающему, что с определенной периодичностью будет удваиваться и число транзисторов на кристалле, и число ядер, при этом сложность ядер и тактовая частота останутся практически неизменными. Использование MapReduce позволит реализовывать архитектуры, попадающие в категорию SPMD (Single Program Multiple Data).

Одним из самых важных моментов предстоящей революции станет отказ от нынешней последовательной модели компьютинга, основанной на исходной редакции схемы фон Неймана, и переход к параллельной. Будет происходить постепенный отказ от SISD (Single Instruction Single Data – «один поток команд, один поток SPMD данных») и переход к SIMD (Single Instruction Multiple Data – «один поток команд, много потоков данных») и MIMD (Multiple Instruction Multiple Data – «много потоков команд, много потоков данных»). Такая классификация, делящая пространство возможностей на четыре части SISD, SIMD, MISD, Multiple Instruction Single Data – «много потоков команд, один поток данных», и MIMD – была предложена Майклом Флинном в начале 70-х годов, когда распараллеливание казалось возможным только на уровне машинных инструкций. Сегодня, с появлением компактных серверов и многоядерных процессоров можно, выполнять параллельно не только инструкции, но и фрагменты кодов, и она может быть уточ-нена – SIMD и MIMD превращаются в SPMD и MPMD (Multiple Program Multiple Data), в том и в другом случае инструкцию заменила программа.

Большинство современных параллельных компьютеров, том числе подавляющая часть супер компьютеров, входящих в список Tор500, принадлежит к категории MPMD. Каждый узел выполняет фрагмент общего задания, работая со своими собственными данными, а в дополнение существует сложная система обмена сообщениями, обеспечивающая согласование совместной работы. Такой вид параллелизма раньше называли параллелизмом на уровне команд (instruction-level parallelism, ILP), а сейчас параллелизмом на уровне задач (task level parallelism), но иногда его еще называют функциональным параллелизмом или параллелизмом по управлению. Его реализация сводится к распределению заданий по узлам и обеспечению синхронности происходящих процессов.

Альтернативный тип параллелизма называют параллелизмом данных (data parallelism), который по классификации Флинна попадает в категорию SIMD. Идея SIMD впервые была реализована еще в 70-е годы на компьютерах CDC Star-100 и Texas Instruments ASC, а популярность эта архитектура приобрела благодаря усилиям Сеймура Крея, особенно успешным оказался компьютер Cray X-MP, который называли векторным процессором за способность работать с одномерными массивами. Реализация SPMD предполагает, что сначала данные должны быть каким-то образом распределены между процессорами, обработаны, а затем собраны. Эту совокупность операций можно назвать map-reduce, как принято в функциональном программировании, хотя точнее было бы split-aggregate, то есть разбиение и сборка, но привилось первое. Возможно несколько способов реализации SPMD: языки высокого уровня, директивы типа loop directives и библиотеки OpenMP, что удобнее на многопроцессорных конфигурациях с общей памятью. На кластерах типа Beowulf можно использовать обмен сообщениями и библиотеку MPI.

Реализация SPMD требует выделения ведущей части кода, ее называют Master или Manager, и подчиненных ей частей Worker. Мастер «раздает» задания Рабочим и потом их собирает. До появления MapReduce эта модель рассматривалась как малоперспективная из-за наличия «бутылочного горла» на тракте обмена между множеством Рабочих и одним Мастером. Создание MapReduce разрешило эту проблему и открыло возможность для обработки огромных массивов данных с использованием архитектуры SPMD.

Допустим, что следует решить простейшую задачу: обработать массив, разбив его на подмассивы. В таком случае работа Мастера сводится к тому, что он делит этот массив на части, посылает каждому из Рабочих положенный ему подмассив, а затем получает результаты и объединяет их (рис.2). В функцию Рабочего входит получение данных, обработка и возврат результатов Мастеру. Распределение нагрузок может быть статическим (static load balancing) или динамическим (dynamic load balancing). Концепция создана по мотивам комбинации map и reduce в функциональном языке программирования Липс. В нем map использует в качестве входных параметров функцию и набор значений, применяя эту функцию по отношению к каждому из значений, а reduce комбинирует полученные результаты.

Рис. 2. Упрощенная схема потока данных в конструкции MapReduce

Созданная в Google конструкция MapReduce делает примерно то же, но по отношению к гигантским объемам данных. В этом случае map – это функция запроса от пользователя, помещенная в библиотеку MapReduce. Ее работа сводится к выбору входной пары, ее обработке и формированию результата в виде значения и некоторого промежуточного ключа, служащего указателем для reduce. Конструкция MapReduce собирает вместе все значения с одинаковыми промежуточными ключами и передает их в функцию reduce, также написанную пользователем. Эта функция воспринимает промежуточные значения, каким-то образом их собирает и воспроизводит результат, скорее всего выраженный меньшим количеством значений,
чем входное множество.

Теперь свяжем функциональную идею MapReduce со схемой SPMD. Вызовы map распределяются между множеством машин путем деления входного потока данных на М срезов (splits или shard), каждый срез обрабатывается на отдельной машине. Вызовы reduce распределены на R частей, количество которых определяется пользователем.

При вызове функции из библиотеки MapReduce выполняется примерно такая последовательность операций (рис. 3).

1. Входные файлы разбиваются на срезы размером от 16 до 64 Мбайт каждый, и на кластере запускаются копии программы. Одна из них Мастер, а остальные – Рабочие. Всего создается M задач map и R задач reduce. Поиском свободных узлов и назначением на них задач занимается Мастер.

2. В процессе исполнения Рабочие, назначенные на выполнение задачи map, считывают содержимое соответствующих срезов, осуществляют их грамматический разбор, выделяют отдельные пары «ключ и соответствующее ему значение», а потом передают эти пары в обрабатывающую их функцию map. Промежуточное значение в виде идентификатора и значения буферизуется в памяти.

3. Периодически буферизованные пары сбрасываются на локальный диск, разделенный на R областей. Расположение этих пар передается Мастеру, который отвечает за дальнейшую передачу этих адресов Рабочим, выполняющим reduce.

4. Рабочие, выполняющие задачу reduce, ждут сообщения от Мастера о местоположении промежуточных пар. По его получении они, используя процедуры удаленного вызова, считывают буферизованные данные с локальных дисков тех Рабочих, которые выполняют map. Загрузив все промежуточные данные, Рабочий сортирует их по промежуточным ключам и, если есть необходимость, группирует.

5. Рабочий обрабатывает данные по промежуточным ключам и передает их в соответствующую функцию reduce для вывода результатов.

6. Когда все задачи map и reduce завершаются, конструкция MapReduce возобновляет работу вызывающей программы и та продолжает выполнять пользовательский код.

Рис. 3. Ход выполнения вызова MapReduce (цифры указывают порядок выполнения)

Одно из важнейших преимуществ этой, замысловатой на первый взгляд конструкции состоит в том, что она позволяет надежно работать на платформах с низкими показателями надежности. Для обнаружения потенциальных сбоев Мастер периодически опрашивает Рабочих, и, если какой-то из них задерживается с ответом сверх заданного норматива, Мастер считает его дефектным и передает исполнение на свободные узлы. Различие между задачами map и reduce в данном случае состоит в том, что map хранит промежуточные результаты на своем диске, и выход из строя такого узла приводит к их потере и требует повторного запуска, а reduce хранит свои данные в глобальном хранилище.

СУБД Greenplum Database

СУБД Greenplum Database 3.2 отличает возможность масштабирования до петабайтных значений с линейным ростом затрат, распараллеливание запросов, ускоряющее скорость работы на один-два порядка, и ряд других эксплуатационных преимуществ. С технологической точки зрения эту СУБД отличает использование программной структуры MapReduce, разработанной в Google, которая обеспечивает не только высокую производительность за счет применения большого числа процессоров, а в перспективе и ядер, но и высокую надежность, а также технологии компрессии, от 3 до 10 раз сокращающей объемы передаваемых и хранимых данных. Greenplum Database 3.2 имеет встроенные возможности для аналитики и статистической обработки средствами специализированного языка программирования R.

Пользовательские функции map и reduce могут быть написаны на различных скриптовых языках, например на Python и Perl: используя привычные средства, разработчик получает доступ к файлам и Web-сайтам непосредственно по месту их нахождения (where it lives), причем делается это проще и естественнее, чем при использовании технологий, связанных с реляционными СУБД. СУБД Greenplum является одновременно и коммерческой реализацией конструкции MapReduce, обеспечивающей работу с большими документами в разных форматах, и классической СУБД, ориентированной на работу с реляционными таблицами (рис. 4). Такого рода универсальность достигается за счет машины для работы с параллельными потоками данных – СУБД может не только независимо работать с каждым из двух источников, но и совмещать работу с ними двумя одновременно. В частности, средствами MapReduce можно оптимизировать доступ к большим СУБД либо обеспечить выполнение SQL-запросов к таблицам и к файлам.

Рис. 4. Greenplum Database: два источника данных и два метода доступа

Ядро Greenplum Database – Parallel Dataflow Engine, машина для обработки параллельных потоков данных, которая может оптимизировать процессы, собирая данные из внешних источников (файлов и приложений) и работая с локальными дисками или дисками на сегментах, объединенных в общую сеть фирменным межсоединением gNet. Машина задумана так, что в перспективе может поддерживать системы, состоящие из многих тысяч процессорных ядер и при этом поддерживать SQL в характерной для нее параллельной манере.

Аппаратно-программные хранилища

Для поддержки оперативного доступа ко все большим объемам информации требуются новые архитектуры платформ хранилищ данных.

Многоядерные процессоры и грядущая параллельная революция

На протяжении длительного времени прогресс в области микропроцессоров отождествлялся с тактовой частотой, но правы оказались те, кто сделал ставку на многоядерные архитектуры.

Рис. 1. Архитектуры хранилищ данных

Рис. 2. Упрощенная схема потока данных в конструкции MapReduce

Рис. 3. Ход выполнения вызова MapReduce (цифры указывают порядок выполнения)

Рис. 4. Greenplum Database: два источника данных и два метода доступа

MapReduce — модель распределённых вычислений, представленная компанией Google, используемая для параллельных вычислений над очень большими, несколько петабайт,^[1] наборами данных в компьютерных кластерах.

Обзор

MapReduce — это фреймворк для вычисления некоторых наборов распределенных задач с использованием большого количества компьютеров (называемых «нодами»), образующих кластер.

Работа MapReduce состоит из двух шагов: Map и Reduce.

На Map-шаге происходит предварительная обработка входных данных. Для этого один из компьютеров (называемый главным узлом — master node) получает входные данные задачи, разделяет их на части и передает другим компьютерам (рабочим узлам — worker node) для предварительной обработки. Название данный шаг получил от одноименной функции высшего порядка.

На Reduce-шаге происходит свёртка предварительно обработанных данных. Главный узел получает ответы от рабочих узлов и на их основе формирует результат — решение задачи, которая изначально формулировалась.

Преимущество MapReduce заключается в том, что он позволяет распределенно производить операции предварительной обработки и свертки. Операции предварительной обработки работают независимо друг от друга и могут производиться параллельно (хотя на практике это ограничено источником входных данных и/или количеством используемых процессоров). Аналогично, множество рабочих узлов могут осуществлять свертку — для этого необходимо только чтобы все результаты предварительной обработки с одним конкретным значением ключа обрабатывались одним рабочим узлом в один момент времени. Хотя этот процесс может быть менее эффективным по сравнению с более последовательными алгоритмами, MapReduce может быть применен к большим объёмам данных, которые могут обрабатываться большим количеством серверов. Так, MapReduce может быть использован для сортировки петабайта данных, что займет всего лишь несколько часов. Параллелизм также дает некоторые возможности восстановления после частичных сбоев серверов: если в рабочем узле, производящем операцию предварительной обработки или свертки, возникает сбой, то его работа может быть передана другому рабочему узлу (при условии, что входные данные для проводимой операции доступны).

Фреймворк в большой степени основан на функциях map и reduce, широко используемых в функциональном программировании,^[2] хотя фактически семантика фреймворка отличается от прототипа.^[3]

Пример

Канонический пример приложения, написанного с помощью MapReduce — это процесс, подсчитывающий, сколько раз различные слова встречаются в наборе документов:

// Функция, используемая рабочими нодами на Map-шаге
// для обработки пар ключ-значение из входного потока
void map(String name, String document):
    // Входные данные:
    //   name - название документа
    //   document - содержимое документа
    for each word w in document:
        EmitIntermediate(w, "1");
 
// Функция, используемая рабочими нодами на Reduce-шаге
// для обрабоки пар ключ-значение, полученных на Map-шаге
void reduce(String word, Iterator partialCounts):
    // Входные данные:
    //   word - слово
    //   partialCounts - список группированных промежуточных результатов. Количество записей в partialCounts и есть 
    //     требуемое значение
    int result = 0;
    for each v in partialCounts:
        result += parseInt(v);
    Emit(AsString(result));

В этом коде на Map-шаге каждый документ разбивается на слова, и возвращаются пары, где ключом является само слово, а значением — «1». Если в документе одно и то же слово встречается несколько раз, то в результате предварительной обработки этого документа будет столько же этих пар, сколько раз встретилось это слово.

Библиотека объединяет все пары с одинаковым ключом и передает их на вход функции reduce, которой остается сложить их, чтобы получить общее количество вхождений данного слова во все документы.

Имеющиеся реализации

• Google реализовал MapReduce на C++ с интерфейсами на языках Python и Java
• Greenplum — коммерческая реализация MapReduce с поддержкой языков Python, Perl, SQL и других.^[4]
• GridGain — бесплатная реализация MapReduce с открытым исходным кодом на языке Java.
• Проект Apache Hadoop — бесплатная реализация MapReduce с открытым исходным кодом на языке Java
• Phoenix [1] — реализация MapReduce на языке С с использованием разделяемой памяти
• MapReduce также реализована Cell Broadband Engine на языке C [2]
• MapReduce реализована в графических процессорах NVIDIA с использованием CUDA [3].
• Qt Concurrent — упрощённая версия фреймворка, реализованная на C++, которая используется для распределения задачи между несколькими ядрами одного компьютера.
• CouchDB использует MapReduce для определения представлений поверх распределённых документов
• MongoDB позволяет использовать MapReduce для параллельной обработки запросов на нескольких серверах
• Skynet — реализация с открытым исходным кодом на языке Ruby
• Disco — реализация MapReduce, созданная компанией Nokia. Её ядро написано на языке Erlang, а приложения для неё можно писать на языке Python
• Hive framework — надстройка с открытым исходным кодом от Facebook, позволяющая комбинировать подход MapReduce и доступ к данным на SQL-подобном языке.
• Qizmt — реализация MapReduce с открытым исходным кодом от MySpace, написанная на C#.
• DryadLINQ — реализация MapReduce, созданная подразделением Microsoft Research в том числе на основе PLINQ и Dryad.

См. также

• Apache Hadoop
• Google File System

Примечания

Ссылки

• MapReduce и параллельные СУБД: друзья или враги?, citforum.ru
• IBM MapReduce Tools for Eclipse

Google Inc.
Председатель совета директоров: Эрик Шмидт • Директор, президент по технологиям и сооснователь: Сергей Брин • Главный исполнительный директор и сооснователь: Ларри Пейдж
Реклама	Adscape • AdSense • Advertising Professional • AdWords • Analytics • Checkout • Click-to-Call • DoubleClick • Insights for Search • Trends • Wallet • Google Университет
Коммуникации	Google+ • Answers • Buzz • Calendar • Dodgeball • Friend Connect • Gmail (история • интерфейс) • Groups • Joga Bonito • Orkut • Panoramio • Picasa • Вопросы и ответы • Reader • Talk • Translate • Voice • Wave
ПО	Browser Sync • Chrome • Chromium • Desktop • Earth • Gadgets • Goggles • Lively • Gmail Mobile • Pack • Picasa • Picnik • SketchUp • Talk • Toolbar • Updater • Web Accelerator
Платформы	Account • Android • App Engine • Apps • Base • BigTable • Chrome OS • Co-op • Gears • Native Client • GFS • Health • Mashup • OpenSocial
Разрабатываемые инструменты	Code • Dart • Gadgets API • GData • Go • Googlebot • Guice • GWS • Highly Open Participation Contest • Image Labeler • KML • MapReduce • Mediabot • Pinyin • SketchUp Ruby • Sitemaps (index) • Summer of Code • TechTalks • Web Toolkit • Website Optimizer
Публикация	Alerts • Blogger • Bookmarks • Docs • FeedBurner • iGoogle • Jaiku • Knol • Library Project • Map Maker • Mashup Editor • Notebook • Page Creator • Sites • Video Marketplace • YouTube • Диск
Поиск (PageRank, руководства)	Appliance • Audio • Books • Code • Desktop • GOOG-411 • Images • Maps (Mars • Moon • Ocean • Sky • Street View) • News • Patents • Products • Scholar • SearchWiki • Usenet • Video • Web
Тематические проекты	Finance
См. также	Поглощения • Цензура • Current • Критика • Earth Outreach • Foundation (Google.org) • Google China • Googleplex • История • Hoaxes • I’m Feeling Lucky • Labs • Logo • Lunar X Prize • I/O • Список сервисов и инструментов • WiFi • Zeitgeist • «Бомбы» • Гуглизм

Уровень сложности
Простой

Время на прочтение
3 мин

Количество просмотров 6.6K

Здраствуйте читатели)

Перед тем как начать, хочу отметить, что данная статья служит для любопытствующих людей и тем, кому нужно базовое теоретическое знание MapReduce’а. Сам по себе MapReduce уже устарел. Если вы в поиске хороших решений, то увы, в этой статье не будет вестись речь о готовых инструментах. О них вы можете почитать в комментариях или написать свой, буду благодарен)

(P.S: Спасибо за замечания, интерес и активность Revertis, procfg, Stas911 и другим^^)

Пример задачи

Хотим автоматизировать огромный фруктовый рынок. На каждое событие будем писать строчку в структурированный лог.

Этот лог не является частью runtime функционирования рынка, но может быть полезен для изучения статистики и аналитики. Например, на основании лога продавец может сделать вывод, что свежие яблоки выгоднее привозить к 13:00.

Хотим узнать

Сколько всего денег покупатели потратили на каждый тип фруктов?

Почему бы не реляционная бд + SQL?

Пусть все эти данные лежат в таблице log в MySQL и мы просто делаем такой запрос:

select fruit, sum(price)
from log
where event = 'buy'
group by fruit;

Это будет работать. Но наш рынок огромный! И размер этого лога 100 петабайт. Он не поместится на жесткий диск одной машины.

А что если в Hadoop?

Пусть этот лог пишется в hadoop кластер. Hadoop сам разложит эту «таблицу» на разные серверы (обычно называется «ноды» node).

Например так:

За распределение данных по кластеру отвечает HDFS (Hadoop Distributed File System). Из-за того, что данные разделены на разные ноды, объем данных не ограничен размером жесткого диска одной ноды. А репликация обеспечивает отказоустойчивость (если одна из нод сломается, реплика встанет на ее место).

Теперь с помощью двух операций map и reduce получим ответ на наш вопрос.

Пишем map

Операция map очень похожа на функцию map из функционального программирования: она преобразовывает каждую строчку. Но в добавок она может еще и выфильтровать ненужные строчки.

У map из ФП сигнатура такая, например:

Row map(Row row)

Она принимает некий объект класса Row, который представляет строку, и возвращает другой объект Row, который может иметь совсем другой формат. Однако, чтобы такая функция могла еще и фильтровать, ее нужно немного усложнить. Добавим параметр context, в который будем писать строчку вместо того, чтобы ее возвращать (но только если это необходимо):

void map(Row row, Context context)

Теперь напишем имплементацию этой функции, которая:
1. Выбрасывает все события, кроме buy
2. Выбрасывает все столбцы, кроме fruit и price

void map(Row row, Context context) {
    if (row.get("event").equals("buy")) {
        context.write(new Row(
            "fruit", row.get("fruit"),
	    "price", row.get("price")
        ));
    }
}

Это функция будет вызываться для каждой строчки нашей таблицы. Результат map:

Мы получили новую (временную) таблицу.
Теперь должна произойти магия.

Как работает Shuffle?

Этот шаг произведет сам hadoop. Мы только должны сказать ему, какой столбец будет ключом. В нашем случае ключом будет столбец fruit.

Ноды будут обмениваться строчками так, чтобы
все строчки с одинаковым ключом попали на одну ноду

Результат будет такой:

Пишем reduce

Теперь, когда мы уверены, что все бананы лежат на одной и той же ноде (ровно как и другие фрукты), мы можем посчитать их стоимость.

Функция reduce, совсем не отличается от функции reduce в ФП, однако для единообразия возвращать будем так же как в map — через context.

void reduce(List<Row> rows, Context context);

Функция принимает список строчек с одинаковым ключом. И выполняется для каждой такой группы.
Реализация должна просто посчитать сумму:

void reduce(List<Row> rows, Context context) {
    int sum = 0;
    for (Row row : rows) {
        sum += row.get("price");
    }
    context.write(new Row(
        "fruit", rows.get(0).get("fruit"),
        "price", sum
    ));
}

Такой получим результат:

Если вычитать из HDFS новую (временную) таблицу целиком, то получим точно такой же результат, как и с SQL:

Выводы

1. С помощью MapReduce можно сделать все, что можно сделать с SQL.

2. MapReduce сложнее принять, но он изначально нацелен на масштабирование — в кластере могут быть десятки тысяч нод, а твой запрос будет работать почти также, как и с десятком нод.

3. Если данных не так много, то лучше использовать твою любимую реляционную БД, потому что это намного проще и быстрее.

P.S, Старался максимально упростить материал, поэтому в некоторых местах могут быть маленькие обманчики, которые помогают легче понять концепцию.