pax_global_header00006660000000000000000000000064126362744340014525gustar00rootroot0000000000000052 comment=767321e190693437f46610acf86fc48967f2e29b lua-mapreduce-0.4.0/000077500000000000000000000000001263627443400142525ustar00rootroot00000000000000lua-mapreduce-0.4.0/.dir-locals.el000066400000000000000000000001701263627443400167010ustar00rootroot00000000000000( (nil . ( (c-basic-offset . 2) (tab-width . 8) (indent-tabs-mode . nil) ) ) ) lua-mapreduce-0.4.0/.gitignore000066400000000000000000000003741263627443400162460ustar00rootroot00000000000000gmon.out # Building files # ################## *.o *.so # Emacs backups # ################# *~ \#*\# .\#* # Vim swap # *.swp # OS generated files # ###################### .DS_Store .DS_Store? ._* .Spotlight-V100 .Trashes Icon? ehthumbs.db Thumbs.db lua-mapreduce-0.4.0/.gitmodules000066400000000000000000000001461263627443400164300ustar00rootroot00000000000000[submodule "external/luamongo"] path = external/luamongo url = https://github.com/moai/luamongo.git lua-mapreduce-0.4.0/.travis.yml000066400000000000000000000015061263627443400163650ustar00rootroot00000000000000services: - mongodb env: LUA_PATH="./?/init.lua;./?.lua" before_install: - sudo apt-get update -qq - sudo apt-get install -y mongodb mongodb-server screen lua5.2 liblua5.2-dev libboost-filesystem-dev libboost-filesystem-dev libboost-thread-dev libboost-regex-dev libssl-dev - git submodule update --init --recursive - cd external/luamongo/external/mongo-cxx-driver/ && sudo scons --sharedclient=yes --prefix=/usr install && cd - - cd external/luamongo/ && make && sudo mkdir -p /usr/lib/lua/5.2 && sudo cp mongo.so /usr/lib/lua/5.2 && cd - - ssh-keygen -b 2048 -f /home/travis/.ssh/id_rsa -t rsa -q -N "" - ssh-keyscan -t rsa localhost > ~/.ssh/known_hosts - ssh-keyscan -t rsa $(hostname) >> ~/.ssh/known_hosts - cp ~/.ssh/id_rsa.pub ~/.ssh/authorized_keys before_script: - sleep 15 language: lua script: ./test.sh lua-mapreduce-0.4.0/LICENSE000066400000000000000000001045641263627443400152710ustar00rootroot00000000000000Lua-MapReduce, Copyright (c) 2014-2015, Francisco Zamora-Martinez GNU GENERAL PUBLIC LICENSE Version 3, 29 June 2007 Copyright (C) 2007 Free Software Foundation, Inc. Everyone is permitted to copy and distribute verbatim copies of this license document, but changing it is not allowed. Preamble The GNU General Public License is a free, copyleft license for software and other kinds of works. The licenses for most software and other practical works are designed to take away your freedom to share and change the works. By contrast, the GNU General Public License is intended to guarantee your freedom to share and change all versions of a program--to make sure it remains free software for all its users. We, the Free Software Foundation, use the GNU General Public License for most of our software; it applies also to any other work released this way by its authors. You can apply it to your programs, too. When we speak of free software, we are referring to freedom, not price. Our General Public Licenses are designed to make sure that you have the freedom to distribute copies of free software (and charge for them if you wish), that you receive source code or can get it if you want it, that you can change the software or use pieces of it in new free programs, and that you know you can do these things. To protect your rights, we need to prevent others from denying you these rights or asking you to surrender the rights. Therefore, you have certain responsibilities if you distribute copies of the software, or if you modify it: responsibilities to respect the freedom of others. For example, if you distribute copies of such a program, whether gratis or for a fee, you must pass on to the recipients the same freedoms that you received. You must make sure that they, too, receive or can get the source code. And you must show them these terms so they know their rights. Developers that use the GNU GPL protect your rights with two steps: (1) assert copyright on the software, and (2) offer you this License giving you legal permission to copy, distribute and/or modify it. For the developers' and authors' protection, the GPL clearly explains that there is no warranty for this free software. For both users' and authors' sake, the GPL requires that modified versions be marked as changed, so that their problems will not be attributed erroneously to authors of previous versions. Some devices are designed to deny users access to install or run modified versions of the software inside them, although the manufacturer can do so. This is fundamentally incompatible with the aim of protecting users' freedom to change the software. The systematic pattern of such abuse occurs in the area of products for individuals to use, which is precisely where it is most unacceptable. Therefore, we have designed this version of the GPL to prohibit the practice for those products. If such problems arise substantially in other domains, we stand ready to extend this provision to those domains in future versions of the GPL, as needed to protect the freedom of users. Finally, every program is threatened constantly by software patents. States should not allow patents to restrict development and use of software on general-purpose computers, but in those that do, we wish to avoid the special danger that patents applied to a free program could make it effectively proprietary. To prevent this, the GPL assures that patents cannot be used to render the program non-free. The precise terms and conditions for copying, distribution and modification follow. TERMS AND CONDITIONS 0. Definitions. "This License" refers to version 3 of the GNU General Public License. "Copyright" also means copyright-like laws that apply to other kinds of works, such as semiconductor masks. "The Program" refers to any copyrightable work licensed under this License. Each licensee is addressed as "you". "Licensees" and "recipients" may be individuals or organizations. To "modify" a work means to copy from or adapt all or part of the work in a fashion requiring copyright permission, other than the making of an exact copy. The resulting work is called a "modified version" of the earlier work or a work "based on" the earlier work. A "covered work" means either the unmodified Program or a work based on the Program. To "propagate" a work means to do anything with it that, without permission, would make you directly or secondarily liable for infringement under applicable copyright law, except executing it on a computer or modifying a private copy. Propagation includes copying, distribution (with or without modification), making available to the public, and in some countries other activities as well. To "convey" a work means any kind of propagation that enables other parties to make or receive copies. Mere interaction with a user through a computer network, with no transfer of a copy, is not conveying. An interactive user interface displays "Appropriate Legal Notices" to the extent that it includes a convenient and prominently visible feature that (1) displays an appropriate copyright notice, and (2) tells the user that there is no warranty for the work (except to the extent that warranties are provided), that licensees may convey the work under this License, and how to view a copy of this License. If the interface presents a list of user commands or options, such as a menu, a prominent item in the list meets this criterion. 1. Source Code. The "source code" for a work means the preferred form of the work for making modifications to it. "Object code" means any non-source form of a work. A "Standard Interface" means an interface that either is an official standard defined by a recognized standards body, or, in the case of interfaces specified for a particular programming language, one that is widely used among developers working in that language. The "System Libraries" of an executable work include anything, other than the work as a whole, that (a) is included in the normal form of packaging a Major Component, but which is not part of that Major Component, and (b) serves only to enable use of the work with that Major Component, or to implement a Standard Interface for which an implementation is available to the public in source code form. A "Major Component", in this context, means a major essential component (kernel, window system, and so on) of the specific operating system (if any) on which the executable work runs, or a compiler used to produce the work, or an object code interpreter used to run it. The "Corresponding Source" for a work in object code form means all the source code needed to generate, install, and (for an executable work) run the object code and to modify the work, including scripts to control those activities. However, it does not include the work's System Libraries, or general-purpose tools or generally available free programs which are used unmodified in performing those activities but which are not part of the work. For example, Corresponding Source includes interface definition files associated with source files for the work, and the source code for shared libraries and dynamically linked subprograms that the work is specifically designed to require, such as by intimate data communication or control flow between those subprograms and other parts of the work. The Corresponding Source need not include anything that users can regenerate automatically from other parts of the Corresponding Source. The Corresponding Source for a work in source code form is that same work. 2. Basic Permissions. All rights granted under this License are granted for the term of copyright on the Program, and are irrevocable provided the stated conditions are met. This License explicitly affirms your unlimited permission to run the unmodified Program. The output from running a covered work is covered by this License only if the output, given its content, constitutes a covered work. This License acknowledges your rights of fair use or other equivalent, as provided by copyright law. You may make, run and propagate covered works that you do not convey, without conditions so long as your license otherwise remains in force. You may convey covered works to others for the sole purpose of having them make modifications exclusively for you, or provide you with facilities for running those works, provided that you comply with the terms of this License in conveying all material for which you do not control copyright. Those thus making or running the covered works for you must do so exclusively on your behalf, under your direction and control, on terms that prohibit them from making any copies of your copyrighted material outside their relationship with you. Conveying under any other circumstances is permitted solely under the conditions stated below. Sublicensing is not allowed; section 10 makes it unnecessary. 3. Protecting Users' Legal Rights From Anti-Circumvention Law. No covered work shall be deemed part of an effective technological measure under any applicable law fulfilling obligations under article 11 of the WIPO copyright treaty adopted on 20 December 1996, or similar laws prohibiting or restricting circumvention of such measures. When you convey a covered work, you waive any legal power to forbid circumvention of technological measures to the extent such circumvention is effected by exercising rights under this License with respect to the covered work, and you disclaim any intention to limit operation or modification of the work as a means of enforcing, against the work's users, your or third parties' legal rights to forbid circumvention of technological measures. 4. Conveying Verbatim Copies. You may convey verbatim copies of the Program's source code as you receive it, in any medium, provided that you conspicuously and appropriately publish on each copy an appropriate copyright notice; keep intact all notices stating that this License and any non-permissive terms added in accord with section 7 apply to the code; keep intact all notices of the absence of any warranty; and give all recipients a copy of this License along with the Program. You may charge any price or no price for each copy that you convey, and you may offer support or warranty protection for a fee. 5. Conveying Modified Source Versions. You may convey a work based on the Program, or the modifications to produce it from the Program, in the form of source code under the terms of section 4, provided that you also meet all of these conditions: a) The work must carry prominent notices stating that you modified it, and giving a relevant date. b) The work must carry prominent notices stating that it is released under this License and any conditions added under section 7. This requirement modifies the requirement in section 4 to "keep intact all notices". c) You must license the entire work, as a whole, under this License to anyone who comes into possession of a copy. This License will therefore apply, along with any applicable section 7 additional terms, to the whole of the work, and all its parts, regardless of how they are packaged. This License gives no permission to license the work in any other way, but it does not invalidate such permission if you have separately received it. d) If the work has interactive user interfaces, each must display Appropriate Legal Notices; however, if the Program has interactive interfaces that do not display Appropriate Legal Notices, your work need not make them do so. A compilation of a covered work with other separate and independent works, which are not by their nature extensions of the covered work, and which are not combined with it such as to form a larger program, in or on a volume of a storage or distribution medium, is called an "aggregate" if the compilation and its resulting copyright are not used to limit the access or legal rights of the compilation's users beyond what the individual works permit. Inclusion of a covered work in an aggregate does not cause this License to apply to the other parts of the aggregate. 6. Conveying Non-Source Forms. You may convey a covered work in object code form under the terms of sections 4 and 5, provided that you also convey the machine-readable Corresponding Source under the terms of this License, in one of these ways: a) Convey the object code in, or embodied in, a physical product (including a physical distribution medium), accompanied by the Corresponding Source fixed on a durable physical medium customarily used for software interchange. b) Convey the object code in, or embodied in, a physical product (including a physical distribution medium), accompanied by a written offer, valid for at least three years and valid for as long as you offer spare parts or customer support for that product model, to give anyone who possesses the object code either (1) a copy of the Corresponding Source for all the software in the product that is covered by this License, on a durable physical medium customarily used for software interchange, for a price no more than your reasonable cost of physically performing this conveying of source, or (2) access to copy the Corresponding Source from a network server at no charge. c) Convey individual copies of the object code with a copy of the written offer to provide the Corresponding Source. This alternative is allowed only occasionally and noncommercially, and only if you received the object code with such an offer, in accord with subsection 6b. d) Convey the object code by offering access from a designated place (gratis or for a charge), and offer equivalent access to the Corresponding Source in the same way through the same place at no further charge. You need not require recipients to copy the Corresponding Source along with the object code. If the place to copy the object code is a network server, the Corresponding Source may be on a different server (operated by you or a third party) that supports equivalent copying facilities, provided you maintain clear directions next to the object code saying where to find the Corresponding Source. Regardless of what server hosts the Corresponding Source, you remain obligated to ensure that it is available for as long as needed to satisfy these requirements. e) Convey the object code using peer-to-peer transmission, provided you inform other peers where the object code and Corresponding Source of the work are being offered to the general public at no charge under subsection 6d. A separable portion of the object code, whose source code is excluded from the Corresponding Source as a System Library, need not be included in conveying the object code work. A "User Product" is either (1) a "consumer product", which means any tangible personal property which is normally used for personal, family, or household purposes, or (2) anything designed or sold for incorporation into a dwelling. In determining whether a product is a consumer product, doubtful cases shall be resolved in favor of coverage. For a particular product received by a particular user, "normally used" refers to a typical or common use of that class of product, regardless of the status of the particular user or of the way in which the particular user actually uses, or expects or is expected to use, the product. A product is a consumer product regardless of whether the product has substantial commercial, industrial or non-consumer uses, unless such uses represent the only significant mode of use of the product. "Installation Information" for a User Product means any methods, procedures, authorization keys, or other information required to install and execute modified versions of a covered work in that User Product from a modified version of its Corresponding Source. The information must suffice to ensure that the continued functioning of the modified object code is in no case prevented or interfered with solely because modification has been made. If you convey an object code work under this section in, or with, or specifically for use in, a User Product, and the conveying occurs as part of a transaction in which the right of possession and use of the User Product is transferred to the recipient in perpetuity or for a fixed term (regardless of how the transaction is characterized), the Corresponding Source conveyed under this section must be accompanied by the Installation Information. But this requirement does not apply if neither you nor any third party retains the ability to install modified object code on the User Product (for example, the work has been installed in ROM). The requirement to provide Installation Information does not include a requirement to continue to provide support service, warranty, or updates for a work that has been modified or installed by the recipient, or for the User Product in which it has been modified or installed. Access to a network may be denied when the modification itself materially and adversely affects the operation of the network or violates the rules and protocols for communication across the network. Corresponding Source conveyed, and Installation Information provided, in accord with this section must be in a format that is publicly documented (and with an implementation available to the public in source code form), and must require no special password or key for unpacking, reading or copying. 7. Additional Terms. "Additional permissions" are terms that supplement the terms of this License by making exceptions from one or more of its conditions. Additional permissions that are applicable to the entire Program shall be treated as though they were included in this License, to the extent that they are valid under applicable law. If additional permissions apply only to part of the Program, that part may be used separately under those permissions, but the entire Program remains governed by this License without regard to the additional permissions. When you convey a copy of a covered work, you may at your option remove any additional permissions from that copy, or from any part of it. (Additional permissions may be written to require their own removal in certain cases when you modify the work.) You may place additional permissions on material, added by you to a covered work, for which you have or can give appropriate copyright permission. Notwithstanding any other provision of this License, for material you add to a covered work, you may (if authorized by the copyright holders of that material) supplement the terms of this License with terms: a) Disclaiming warranty or limiting liability differently from the terms of sections 15 and 16 of this License; or b) Requiring preservation of specified reasonable legal notices or author attributions in that material or in the Appropriate Legal Notices displayed by works containing it; or c) Prohibiting misrepresentation of the origin of that material, or requiring that modified versions of such material be marked in reasonable ways as different from the original version; or d) Limiting the use for publicity purposes of names of licensors or authors of the material; or e) Declining to grant rights under trademark law for use of some trade names, trademarks, or service marks; or f) Requiring indemnification of licensors and authors of that material by anyone who conveys the material (or modified versions of it) with contractual assumptions of liability to the recipient, for any liability that these contractual assumptions directly impose on those licensors and authors. All other non-permissive additional terms are considered "further restrictions" within the meaning of section 10. If the Program as you received it, or any part of it, contains a notice stating that it is governed by this License along with a term that is a further restriction, you may remove that term. If a license document contains a further restriction but permits relicensing or conveying under this License, you may add to a covered work material governed by the terms of that license document, provided that the further restriction does not survive such relicensing or conveying. If you add terms to a covered work in accord with this section, you must place, in the relevant source files, a statement of the additional terms that apply to those files, or a notice indicating where to find the applicable terms. Additional terms, permissive or non-permissive, may be stated in the form of a separately written license, or stated as exceptions; the above requirements apply either way. 8. Termination. You may not propagate or modify a covered work except as expressly provided under this License. Any attempt otherwise to propagate or modify it is void, and will automatically terminate your rights under this License (including any patent licenses granted under the third paragraph of section 11). However, if you cease all violation of this License, then your license from a particular copyright holder is reinstated (a) provisionally, unless and until the copyright holder explicitly and finally terminates your license, and (b) permanently, if the copyright holder fails to notify you of the violation by some reasonable means prior to 60 days after the cessation. Moreover, your license from a particular copyright holder is reinstated permanently if the copyright holder notifies you of the violation by some reasonable means, this is the first time you have received notice of violation of this License (for any work) from that copyright holder, and you cure the violation prior to 30 days after your receipt of the notice. Termination of your rights under this section does not terminate the licenses of parties who have received copies or rights from you under this License. If your rights have been terminated and not permanently reinstated, you do not qualify to receive new licenses for the same material under section 10. 9. Acceptance Not Required for Having Copies. You are not required to accept this License in order to receive or run a copy of the Program. Ancillary propagation of a covered work occurring solely as a consequence of using peer-to-peer transmission to receive a copy likewise does not require acceptance. However, nothing other than this License grants you permission to propagate or modify any covered work. These actions infringe copyright if you do not accept this License. Therefore, by modifying or propagating a covered work, you indicate your acceptance of this License to do so. 10. Automatic Licensing of Downstream Recipients. Each time you convey a covered work, the recipient automatically receives a license from the original licensors, to run, modify and propagate that work, subject to this License. You are not responsible for enforcing compliance by third parties with this License. An "entity transaction" is a transaction transferring control of an organization, or substantially all assets of one, or subdividing an organization, or merging organizations. If propagation of a covered work results from an entity transaction, each party to that transaction who receives a copy of the work also receives whatever licenses to the work the party's predecessor in interest had or could give under the previous paragraph, plus a right to possession of the Corresponding Source of the work from the predecessor in interest, if the predecessor has it or can get it with reasonable efforts. You may not impose any further restrictions on the exercise of the rights granted or affirmed under this License. For example, you may not impose a license fee, royalty, or other charge for exercise of rights granted under this License, and you may not initiate litigation (including a cross-claim or counterclaim in a lawsuit) alleging that any patent claim is infringed by making, using, selling, offering for sale, or importing the Program or any portion of it. 11. Patents. A "contributor" is a copyright holder who authorizes use under this License of the Program or a work on which the Program is based. The work thus licensed is called the contributor's "contributor version". A contributor's "essential patent claims" are all patent claims owned or controlled by the contributor, whether already acquired or hereafter acquired, that would be infringed by some manner, permitted by this License, of making, using, or selling its contributor version, but do not include claims that would be infringed only as a consequence of further modification of the contributor version. For purposes of this definition, "control" includes the right to grant patent sublicenses in a manner consistent with the requirements of this License. Each contributor grants you a non-exclusive, worldwide, royalty-free patent license under the contributor's essential patent claims, to make, use, sell, offer for sale, import and otherwise run, modify and propagate the contents of its contributor version. In the following three paragraphs, a "patent license" is any express agreement or commitment, however denominated, not to enforce a patent (such as an express permission to practice a patent or covenant not to sue for patent infringement). To "grant" such a patent license to a party means to make such an agreement or commitment not to enforce a patent against the party. If you convey a covered work, knowingly relying on a patent license, and the Corresponding Source of the work is not available for anyone to copy, free of charge and under the terms of this License, through a publicly available network server or other readily accessible means, then you must either (1) cause the Corresponding Source to be so available, or (2) arrange to deprive yourself of the benefit of the patent license for this particular work, or (3) arrange, in a manner consistent with the requirements of this License, to extend the patent license to downstream recipients. "Knowingly relying" means you have actual knowledge that, but for the patent license, your conveying the covered work in a country, or your recipient's use of the covered work in a country, would infringe one or more identifiable patents in that country that you have reason to believe are valid. If, pursuant to or in connection with a single transaction or arrangement, you convey, or propagate by procuring conveyance of, a covered work, and grant a patent license to some of the parties receiving the covered work authorizing them to use, propagate, modify or convey a specific copy of the covered work, then the patent license you grant is automatically extended to all recipients of the covered work and works based on it. A patent license is "discriminatory" if it does not include within the scope of its coverage, prohibits the exercise of, or is conditioned on the non-exercise of one or more of the rights that are specifically granted under this License. You may not convey a covered work if you are a party to an arrangement with a third party that is in the business of distributing software, under which you make payment to the third party based on the extent of your activity of conveying the work, and under which the third party grants, to any of the parties who would receive the covered work from you, a discriminatory patent license (a) in connection with copies of the covered work conveyed by you (or copies made from those copies), or (b) primarily for and in connection with specific products or compilations that contain the covered work, unless you entered into that arrangement, or that patent license was granted, prior to 28 March 2007. Nothing in this License shall be construed as excluding or limiting any implied license or other defenses to infringement that may otherwise be available to you under applicable patent law. 12. No Surrender of Others' Freedom. If conditions are imposed on you (whether by court order, agreement or otherwise) that contradict the conditions of this License, they do not excuse you from the conditions of this License. If you cannot convey a covered work so as to satisfy simultaneously your obligations under this License and any other pertinent obligations, then as a consequence you may not convey it at all. For example, if you agree to terms that obligate you to collect a royalty for further conveying from those to whom you convey the Program, the only way you could satisfy both those terms and this License would be to refrain entirely from conveying the Program. 13. Use with the GNU Affero General Public License. Notwithstanding any other provision of this License, you have permission to link or combine any covered work with a work licensed under version 3 of the GNU Affero General Public License into a single combined work, and to convey the resulting work. The terms of this License will continue to apply to the part which is the covered work, but the special requirements of the GNU Affero General Public License, section 13, concerning interaction through a network will apply to the combination as such. 14. Revised Versions of this License. The Free Software Foundation may publish revised and/or new versions of the GNU General Public License from time to time. Such new versions will be similar in spirit to the present version, but may differ in detail to address new problems or concerns. Each version is given a distinguishing version number. If the Program specifies that a certain numbered version of the GNU General Public License "or any later version" applies to it, you have the option of following the terms and conditions either of that numbered version or of any later version published by the Free Software Foundation. If the Program does not specify a version number of the GNU General Public License, you may choose any version ever published by the Free Software Foundation. If the Program specifies that a proxy can decide which future versions of the GNU General Public License can be used, that proxy's public statement of acceptance of a version permanently authorizes you to choose that version for the Program. Later license versions may give you additional or different permissions. However, no additional obligations are imposed on any author or copyright holder as a result of your choosing to follow a later version. 15. Disclaimer of Warranty. THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION. 16. Limitation of Liability. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MODIFIES AND/OR CONVEYS THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS), EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. 17. Interpretation of Sections 15 and 16. If the disclaimer of warranty and limitation of liability provided above cannot be given local legal effect according to their terms, reviewing courts shall apply local law that most closely approximates an absolute waiver of all civil liability in connection with the Program, unless a warranty or assumption of liability accompanies a copy of the Program in return for a fee. END OF TERMS AND CONDITIONS How to Apply These Terms to Your New Programs If you develop a new program, and you want it to be of the greatest possible use to the public, the best way to achieve this is to make it free software which everyone can redistribute and change under these terms. To do so, attach the following notices to the program. It is safest to attach them to the start of each source file to most effectively state the exclusion of warranty; and each file should have at least the "copyright" line and a pointer to where the full notice is found. {one line to give the program's name and a brief idea of what it does.} Copyright (C) {year} {name of author} This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . Also add information on how to contact you by electronic and paper mail. If the program does terminal interaction, make it output a short notice like this when it starts in an interactive mode: {project} Copyright (C) {year} {fullname} This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'. This is free software, and you are welcome to redistribute it under certain conditions; type `show c' for details. The hypothetical commands `show w' and `show c' should show the appropriate parts of the General Public License. Of course, your program's commands might be different; for a GUI interface, you would use an "about box". You should also get your employer (if you work as a programmer) or school, if any, to sign a "copyright disclaimer" for the program, if necessary. For more information on this, and how to apply and follow the GNU GPL, see . The GNU General Public License does not permit incorporating your program into proprietary programs. If your program is a subroutine library, you may consider it more useful to permit linking proprietary applications with the library. If this is what you want to do, use the GNU Lesser General Public License instead of this License. But first, please read . lua-mapreduce-0.4.0/README.md000066400000000000000000000077631263627443400155460ustar00rootroot00000000000000Lua-MapReduce ============= [![Build Status](https://travis-ci.org/pakozm/lua-mapreduce.png?branch=master)](https://travis-ci.org/pakozm/lua-mapreduce) Travis CI (master branch) [![Build Status](https://travis-ci.org/pakozm/lua-mapreduce.png?branch=devel)](https://travis-ci.org/pakozm/lua-mapreduce) Travis CI (devel branch) Introduction ------------ Lua MapReduce implementation based in MongoDB. It differs from [rohitjoshi/lua-mapreduce](https://github.com/rohitjoshi/lua-mapreduce) in the basis of the communication between the processes. In order to allow fault tolerancy, and to reduce the communication protocol complexity, this implementation relies on mongoDB. So, all the data is stored at auxiliary mongoDB collections. This software depends in: - [Lua 5.2](http://www.lua.org/) - [luamongo](https://github.com/moai/luamongo/), mongoDB driver for Lua 5.2. Installation ------------ Copy the `mapreduce` directory to a place visible from your `LUA_PATH` environment variable. It is possible to add the active directory by writing in the terminal: ``` $ export LUA_PATH='?.lua;?/init.lua' ``` Documentation ------------- Available at [wiki pages](https://github.com/pakozm/lua-mapreduce/wiki). Performance notes ----------------- Word-count example using [Europarl v7 English data](http://www.statmt.org/europarl/), with *1,965,734 lines* and *49,158,635 running words*. The data has been splitted in 197 files with a maximum of *10,000* lines per file. The task is executed in *one machine* with *four cores*. The machine runs a MongoDB server, a lua-mapreduce server and four lua-mapreduce workers. **Note** that this task is not fair because the data could be stored in the local filesystem. The output of lua-mapreduce was: ``` $ ./execute_BIG_server.sh > output # Iteration 1 # Preparing Map # Map execution, size= 197 100.0 % # Preparing Reduce # Reduce execution, num_files= 1970 size= 10 100.0 % # Map sum(cpu_time) 80.297174 # Reduce sum(cpu_time) 56.829328 # Sum(cpu_time) 137.126502 # Map sum(real_time) 84.476371 # Reduce sum(real_time) 63.458693 # Sum(real_time) 147.935064 # Sum(sys_time) 10.808562 # Map cluster time 26.661836 # Reduce cluster time 20.710385 # Cluster time 47.372221 # Failed maps 0 # Failed reduces 0 # Server time 49.229152 # Final execution ``` **Note 1:** using only one worker takes: 146 seconds **Note 2:** using 30 mappers and 15 reducers (30 workers) takes: 32 seconds A naive word-count version implemented with pipes and shellscripts takes: ``` $ time cat /home/experimentos/CORPORA/EUROPARL/en-splits/* | \ tr ' ' '\n' | sort | uniq -c > output-pipes real 2m21.272s user 2m23.339s sys 0m2.951s ``` A naive word-count version implemented in Lua takes: ``` $ time cat /home/experimentos/CORPORA/EUROPARL/en-splits/* | \ lua misc/naive.lua > output-naivetime real 0m26.125s user 0m17.458s sys 0m0.324s ``` Looking to these numbers, it is clear that the better is to work in main memory and in local storage filesystem, as in the naive Lua implementation, which needs only 26 seconds (real time), but uses local disk files. The map-reduce approach takes 49 seconds (real time) with four workers and 146 seconds (real time) with only one worker. These last two numbers are comparable with the naive shellscript implementation using pipes, which takes 146 seconds (real time). Concluding, the preliminar lua-mapreduce implementation, using four workers and MongoDB for communication and GridFS for auxiliary storage, is up to **3** times faster than a shellscript implementation using pipes. Both implementations sort the data in order to aggregate the results. In the future, a larger data task will be choosen to compare this implementation with raw map-reduce in MongoDB and/or Hadoop. Last notes ---------- This software is in development. More documentation will be added to the wiki pages, while we have time to do that. Collaboration is open, and all your contributions will be welcome. lua-mapreduce-0.4.0/execute_BIG_server.sh000077500000000000000000000005331263627443400203230ustar00rootroot00000000000000#!/bin/bash LUA_PATH="?.lua;?/init.lua" lua execute_server.lua django wordcountBIG \ mapreduce.examples.WordCountBig.taskfn \ mapreduce.examples.WordCount.mapfn \ mapreduce.examples.WordCount.partitionfn \ mapreduce.examples.WordCount.reducefn \ mapreduce.examples.WordCount.finalfn \ mapreduce.examples.WordCount.reducefn lua-mapreduce-0.4.0/execute_BIG_worker.sh000077500000000000000000000001231263627443400203210ustar00rootroot00000000000000#!/bin/bash LUA_PATH="?.lua;?/init.lua" lua execute_worker.lua django wordcountBIG lua-mapreduce-0.4.0/execute_example_server.sh000077500000000000000000000004771263627443400213640ustar00rootroot00000000000000#!/bin/bash lua execute_server.lua localhost wordcount \ mapreduce.examples.WordCount.taskfn \ mapreduce.examples.WordCount.mapfn \ mapreduce.examples.WordCount.partitionfn \ mapreduce.examples.WordCount.reducefn \ mapreduce.examples.WordCount.finalfn \ mapreduce.examples.WordCount.reducefn $@ lua-mapreduce-0.4.0/execute_example_worker.sh000077500000000000000000000000671263627443400213620ustar00rootroot00000000000000#!/bin/bash lua execute_worker.lua localhost wordcount lua-mapreduce-0.4.0/execute_server.lua000066400000000000000000000041651263627443400200130ustar00rootroot00000000000000-- This is a generic server executing script. Its arguments are: -- -- [1] => connection_string -- -- [2] => name of the database -- -- [3] => taskfn Lua module, it could be given as a Lua require string (using -- dots as sep) or as a path (using / as sep) -- -- [4] => mapfn Lua module, idem -- -- [5] => partitionfn Lua module, idem -- -- [6] => reducefn Lua module, idem -- -- [7] => finalfn Lua module, idem (OPTIONAL, by default it is nothing) -- -- [8] => combinerfn Lua module, idem (OPTIONAL, by default it is nothing. It could be nil) -- -- [9] => storage (OPTIONAL, by default is nothing. It could be nil) -- -- IMPORTANT: the Lua modules (taskfn, mapfn, reducefn, ...) need to be in the -- LUA_PATH in all the machines where this code need to be executed -- local connection_string = table.remove(arg, 1) local dbname = table.remove(arg, 1) local taskfn = table.remove(arg, 1) local mapfn = table.remove(arg, 1) local partitionfn = table.remove(arg, 1) local reducefn = table.remove(arg, 1) local finalfn = table.remove(arg, 1) local combinerfn = table.remove(arg, 1) local storage = table.remove(arg, 1) if combinerfn == "nil" then combinerfn = nil end if storage == "nil" then storage = nil end -- local function normalize(name) return name:gsub("/","."):gsub("%.lua$","") end -- local mapreduce = require "mapreduce" local s = mapreduce.server.new(connection_string, dbname) s:configure{ taskfn = normalize(taskfn), mapfn = normalize(mapfn), partitionfn = normalize(partitionfn), reducefn = normalize(reducefn), finalfn = (finalfn and normalize(finalfn)) or nil, combinerfn = (combinerfn and normalize(combinerfn)) or nil, init_args = arg, storage = storage, -- storage = "gridfs[:PATH]", -- 'gridfs', 'shared', 'sshfs', with the -- optional string :PATH. if not given PATH will be os.tmpname() -- storage = "gridfs:/tmp/wordcount", -- storage = "shared:/home/experimentos/tmp/wordcount", -- storage = "sshfs:/tmp/wordcount", -- storage = "gridfs", -- storage = "shared", -- storage = "sshfs", } mapreduce.utils.sleep(4) s:loop() lua-mapreduce-0.4.0/execute_worker.lua000066400000000000000000000004521263627443400200110ustar00rootroot00000000000000-- This is a generic worker executing script. Its arguments are: -- -- [1] => connection_string -- -- [2] => name of the database -- local connection_string = arg[1] local dbname = arg[2] local mapreduce = require "mapreduce" local w = mapreduce.worker.new(connection_string, dbname) w:execute() lua-mapreduce-0.4.0/external/000077500000000000000000000000001263627443400160745ustar00rootroot00000000000000lua-mapreduce-0.4.0/external/luamongo/000077500000000000000000000000001263627443400177155ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/000077500000000000000000000000001263627443400162175ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/cnn.lua000066400000000000000000000126041263627443400175030ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- The cnn class represents an abstraction layer over MongoDB. It allows to -- retrieve connections, gridfs objects, file builders, insertion of errors and -- batch insertion procedure. local mongo = require "mongo" local utils = require "mapreduce.utils" local cnn = { _VERSION = "0.2", _NAME = "cnn", } -- performs the connection, allowing to retrive a lost connection, and returns a -- dbclient object function cnn:connect() if not self.db or self.db:is_failed() then self.db = utils.connect(self.connection_string, self.auth_table) end return self.db end function cnn:gridfs() local db = self:connect() local gridfs = assert( mongo.GridFS.New(db, self.gridfs_dbname) ) return gridfs end -- this weak table allow to index gridfs objects using weak grid_file_builder -- keys local gridfs_instances = setmetatable({}, { __mode="k" }) function cnn:grid_file_builder() local gridfs = self:gridfs() local grid_file_builder = mongo.GridFileBuilder.New(gridfs) gridfs_instances[grid_file_builder] = gridfs return grid_file_builder end function cnn:get_dbname() return self.dbname end function cnn:insert_error(who,msg) local ns = string.format("%s.errors", self.dbname) local db = self:connect() db:insert(ns, { worker = who, msg = msg }) end function cnn:get_errors() local ns = string.format("%s.errors", self.dbname) local db = self:connect() return db:query(ns, {}) end function cnn:remove_errors(ids) local ns = string.format("%s.errors", self.dbname) local db = self:connect() db:remove(ns,{ _id = { ["$in"] = ids } }) end function cnn:annotate_insert(ns,tbl,callback) self.pending_inserts = self.pending_inserts or {} self.pending_callbacks = self.pending_callbacks or {} self.pending_inserts[ns] = self.pending_inserts[ns] or {} self.pending_callbacks[ns] = self.pending_callbacks[ns] or {} table.insert(self.pending_inserts[ns], tbl) if callback then table.insert(self.pending_callbacks[ns], callback) end if #self.pending_inserts[ns] >= utils.MAX_PENDING_INSERTS then local db = self:connect() db:insert_batch(ns,self.pending_inserts[ns]) for i,func in ipairs(self.pending_callbacks[ns]) do func(self.pending_inserts[ns][i]) end self.pending_inserts[ns] = {} self.pending_callbacks[ns] = {} end end function cnn:flush_pending_inserts(max) local max = max or 0 if self.pending_inserts then local db = self:connect() for ns,tbl in pairs(self.pending_inserts) do if #tbl > max then db:insert_batch(ns,tbl) for i,func in ipairs(self.pending_callbacks[ns]) do func(tbl[i]) end end end self.pending_inserts = nil self.pending_callbacks = nil end end function cnn:__call(connection_string, dbname, auth_table) local obj = { connection_string = connection_string, dbname = dbname, gridfs_dbname = dbname, auth_table = auth_table } setmetatable(obj, { __index=self }) return obj end setmetatable(cnn,cnn) ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- cnn.utest = function() local c = cnn("localhost","test") local db = assert( c:connect() ) assert( not db:is_failed() ) local gridfs = assert( c:gridfs() ) local builder = assert( c:grid_file_builder() ) assert(c:get_dbname() == "test") -- insert error db:drop_collection("test.errors") for _,who in ipairs{ "utest1", "utest2" } do for _,msg in ipairs{ "error 1", "error 2" } do c:insert_error(who, msg) end end assert(db:count("test.errors") == 4) -- get errors local ids = {} local q = c:get_errors() for r in q:results() do assert(r.worker == "utest1" or r.worker == "utest2") assert(r.msg == "error 1" or r.msg == "error 2") table.insert(ids, r._id) end -- remove errors c:remove_errors(ids) assert(db:count("test.errors") == 0) -- annotate insert db:drop_collection("test.inserts") local insert_results = {} for i=1,utils.MAX_PENDING_INSERTS+10 do c:annotate_insert("test.inserts", { _id=i, value=utils.time() }, function(v) assert(v._id == i) table.insert(insert_results, i) end) end assert(#insert_results == utils.MAX_PENDING_INSERTS) assert(db:count("test.inserts") == utils.MAX_PENDING_INSERTS) c:flush_pending_inserts(0) assert(db:count("test.inserts") == utils.MAX_PENDING_INSERTS+10) assert(#insert_results == utils.MAX_PENDING_INSERTS+10) for i=1,#insert_results do assert(insert_results[i] == i) end end return cnn lua-mapreduce-0.4.0/mapreduce/examples/000077500000000000000000000000001263627443400200355ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/examples/APRIL-ANN/000077500000000000000000000000001263627443400213165ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/examples/APRIL-ANN/common.lua000066400000000000000000000201071263627443400233110ustar00rootroot00000000000000-- Copyright 2014, Francisco Zamora-Martinez -- -- Common functions for APRIL-ANN training scripts -- https://github.com/pakozm/april-ann local mongo = require "mongo" local mr_utils = require "mapreduce.utils" local ptable = require "mapreduce.persistent_table" ----------------------------------------------------------------------------- local TR_LOSS_KEY = "TR_LOSS" local NUM_REDUCERS local dbname,dbhost local db,gridfs,conf,params local user_init,user_taskfn,user_finalfn local generate_new_trainer,compute_gradients_and_loss,compute_validation_loss ----------------------------------------------------------------------------- util.omp_set_num_threads(1) ----------------------------------------------------------------------------- local serialize_to_gridfs = function(gridfs, filename, obj) gridfs:remove_file(filename) local builder = mongo.GridFileBuilder.New(gridfs) builder:append(util.serialize(obj)) builder:build(filename) end local deserialize_from_gridfs = function(gridfs, filename) local file = assert( gridfs:find_file_by_name(filename) ) local str_tbl = {} for i=1,file:num_chunks() do local chunk = file:chunk(i-1) table.insert(str_tbl, chunk:data()) end return util.deserialize(table.concat(str_tbl)) end local serialize_and_map_emit = function(key,value,emit) emit(key, util.serialize(value)) end local serialize_and_red_emit = function(value,emit) emit(util.serialize(value)) end local deserialize_emitted_value = function(str) return util.deserialize(str) end ------------------------------------------------------------------------------ -- the first init is executed is at MapReduce server instance when it is -- configured local init = function(arg) conf:update() if not conf.train_func or conf.finished then if conf.finished then conf:drop() local list = gridfs:list() for r in list:results() do gridfs:remove_file(r.filename) end end conf.finished = false -- store current iteration and best model in a GridFS file (tmpname) conf.train_func = os.tmpname() -- mr_utils.remove(conf.train_func) local trainer,train_func = generate_new_trainer_and_train_func() train_func:get_state_table().last = trainer serialize_to_gridfs(gridfs, conf.train_func, train_func) conf:update() end user_init(arg, conf) if conf:dirty() then conf:update() end end local taskfn = function(emit) conf:update() user_taskfn(emit,conf) if conf:dirty() then conf:update() end end local mapfn = function(key, value, emit) conf:update() local train_func = deserialize_from_gridfs(gridfs, assert(conf.train_func)) local trainer = train_func:get_state_table().last conf:read_only(true) local weight_grads,loss_matrix,bunch_size = compute_gradients_and_loss(trainer, key, value, conf) conf:read_only(false) assert(weight_grads and loss_matrix and bunch_size, "compute_gradients_and_loss had to return gradients, loss_matrix and bunch_size") for name,grads in pairs(weight_grads) do serialize_and_map_emit(name, { grads, trainer:weights(name):get_shared_count()*bunch_size }, emit) end serialize_and_map_emit(TR_LOSS_KEY, loss_matrix, emit) end local partitionfn = function(key) local sum = 0 for i=1,#key do sum = sum + key:byte(i) end return sum % NUM_REDUCERS end local loss local reducefn = function(key, values, emit) if not loss then local train_func = deserialize_from_gridfs(gridfs, assert(conf.train_func)) local trainer = train_func:get_state_table().last loss = trainer:get_loss_function() end if key == TR_LOSS_KEY then loss:reset() for i=1,#values do local v = deserialize_emitted_value(values[i]) loss:accum_loss(v) end serialize_and_red_emit({ loss:get_accum_loss() }, emit) else -- accumulate gradients and shared count local t = deserialize_emitted_value(values[1]) local gradient = t[1] local counts = t[2] for i=2,#values do local t = deserialize_emitted_value(values[i]) gradient:axpy(1.0, t[1]) counts = counts + t[2] end serialize_and_red_emit({ gradient, counts }, emit) end end local train_func local trainer local optimizer local thenet local loss local finalfn = function(pairs_iterator) local weight_grads = matrix.dict() conf:update() if not train_func then train_func = deserialize_from_gridfs(gridfs, assert(conf.train_func)) trainer = train_func:get_state_table().last optimizer = trainer:get_optimizer() thenet = trainer:get_component() loss = trainer:get_loss_function() end -- local tr_loss_mean,tr_loss_var for key,values in pairs_iterator do local value = deserialize_emitted_value(values[1]) if key == TR_LOSS_KEY then tr_loss_mean = value[1] tr_loss_var = value[2] else local N = value[2] if not N or N==0 then N=1 end if params.smooth_gradients then -- gradients smoothing value[1]:scal( 1.0/math.sqrt(N) ) end weight_grads[key] = value[1] end end assert(tr_loss_mean) -- conf:read_only(true) local weights_table = trainer:get_weights_table() local non_stop = train_func:execute(function() optimizer:execute(function(x,it) if x ~= weights_table then trainer:build{ weights = x } weights_table = x end assert(not it or it == 0) thenet:reset(it) return tr_loss_mean,weight_grads end, weights_table) local va_loss_mean,va_loss_var = compute_validation_loss(trainer, conf) return trainer, tr_loss_mean, va_loss_mean end) conf:read_only(false) serialize_to_gridfs(gridfs, assert(conf.train_func), train_func) -- if non_stop then user_finalfn(train_func, conf) if conf:dirty() then conf:update() end return "loop" else conf.finished = true conf:update() return true end end ------------------------------------------------------------------------------ local make_map_reduce_task_table = function(t) params = get_table_fields( { num_reducers = { mandatory = true, type_match="number" }, dbname = { mandatory = true, type_match="string" }, dbhost = { mandatory = true, type_match="string" }, compute_gradients_and_loss = { mandatory = true, type_match="function" }, compute_validation_loss = { mandatory = true, type_match="function" }, user_init = { mandatory = true, type_match="function" }, user_taskfn = { mandatory = true, type_match="function" }, user_finalfn = { mandatory = true, type_match="function" }, generate_new_trainer_and_train_func = { mandatory = true, type_match="function" }, smooth_gradients = { mandatory = false, default = true }, }, t) -- dbname = params.dbname dbhost = params.dbhost db = mr_utils.connect(dbhost) gridfs = mongo.GridFS.New(db, dbname) -- persistent table allows to store data which will be accessed in a -- distributed way conf = ptable("conf", dbhost, dbname) compute_gradients_and_loss = params.compute_gradients_and_loss compute_validation_loss = params.compute_validation_loss generate_new_trainer_and_train_func = params.generate_new_trainer_and_train_func user_init = params.user_init user_taskfn = params.user_taskfn user_finalfn = params.user_finalfn NUM_REDUCERS = params.num_reducers return { init = init, -- taskfn = taskfn, mapfn = mapfn, partitionfn = partitionfn, reducefn = reducefn, finalfn = finalfn, } end local aux_cache = {} local cached = function(key,func,cache) local cache = cache or aux_cache if not cache[key] then cache[key] = func(key) end return cache[key] end ------------------------------------------------------------------------------ return { make_map_reduce_task_table = make_map_reduce_task_table, cached = cached, serialize_to_gridfs = serialize_to_gridfs, deserialize_from_gridfs = deserialize_from_gridfs, } lua-mapreduce-0.4.0/mapreduce/examples/APRIL-ANN/init.lua000066400000000000000000000155561263627443400230000ustar00rootroot00000000000000-- Copyright 2014, Francisco Zamora-Martinez -- -- Example of iterative MapReduce to train an ANN using APRIL-ANN toolkit: -- https://github.com/pakozm/april-ann local NUM_REDUCERS = 10 local EXP_DBHOST = "localhost" local EXP_DBNAME = "exp_digits" local bunch_size = 128 local weights_random = random(1234) local description = "256 inputs 128 tanh 10 log_softmax" local inf = -1 local sup = 1 local shuffle_random = random() -- TOTALLY RANDOM FOR EACH WORKER local learning_rate = 0.01 local momentum = 0.02 local weight_decay = 1e-04 local max_epochs = 40 local min_epochs = 20 local common = require "mapreduce.examples.APRIL-ANN.common" local cached = common.cached -- local mat_cache = {} local ds_cache = {} -- generates a new MLP with random weights local generate_new_trainer_and_train_func = function() local thenet = ann.mlp.all_all.generate(description) local trainer = trainable.supervised_trainer(thenet, ann.loss.multi_class_cross_entropy(10), bunch_size) trainer:build() trainer:randomize_weights{ random = weights_random, inf = inf, sup = sup, use_fanin = true, } trainer:set_option("learning_rate", learning_rate) trainer:set_option("momentum", momentum) trainer:set_option("weight_decay", weight_decay) -- it is better to avoid BIAS regularization trainer:set_layerwise_option("b.", "weight_decay", 0) -- local stopping_criterion = trainable.stopping_criteria.make_max_epochs_wo_imp_relative(2) local train_func = trainable.train_holdout_validation{ min_epochs = min_epochs, max_epochs = max_epochs, stopping_criterion = stopping_criterion } return trainer,train_func end -- the first init is executed is at MapReduce server instance when it is -- configured; this function receives the arguments list given to server -- configure method, and a persistent table where it is possible to store -- persistent data which could be retrieved from map/reduce functions local user_init = function(arg,conf) end -- receives the emit function and the persistent table in read-only mode local user_taskfn = function(emit,conf) emit(1,"misc/digits.png") emit(2,"misc/digits.png") emit(3,"misc/digits.png") emit(4,"misc/digits.png") end local m2 = matrix(10,{1,0,0,0,0,0,0,0,0,0}) local make_load_matrix = function(value) return function() return ImageIO.read(value):to_grayscale():invert_colors():matrix() end end local make_load_dataset = function(mat,m2) return function() local train_input = dataset.matrix(mat, { patternSize = {16,16}, offset = {0,0}, numSteps = {80,10}, stepSize = {16,16}, orderStep = {1,0} }) -- a circular dataset which advances with step -1 local train_output = dataset.matrix(m2, { patternSize = {10}, offset = {0}, numSteps = {800}, stepSize = {-1}, circular = {true} }) -- VALIDATION -- local val_input = dataset.matrix(mat, { patternSize = {16,16}, offset = {1280,0}, numSteps = {20,10}, stepSize = {16,16}, orderStep = {1,0} }) local val_output = dataset.matrix(m2, { patternSize = {10}, offset = {0}, numSteps = {200}, stepSize = {-1}, circular = {true} }) return { train_input = dataset.token.wrapper(train_input), train_output = dataset.token.wrapper(train_output), val_input = dataset.token.wrapper(val_input), val_output = dataset.token.wrapper(val_output), } end end -- receives a trainer, key,value pair and the persistent table in read-only mode -- as last argument; returns gradients, loss_matrix and bunch_size local compute_gradients_and_loss = function(trainer, key, value, conf) local mat = cached(value, make_load_matrix(value), mat_cache) local ds_tbl = cached(value, make_load_dataset(mat,m2), ds_cache) local in_ds = ds_tbl.train_input local out_ds = ds_tbl.train_output local bunch_tbl = {} for i=1,bunch_size do table.insert(bunch_tbl, shuffle_random:randInt(1,in_ds:numPatterns())) end local input = in_ds:getPatternBunch(bunch_tbl) local target = out_ds:getPatternBunch(bunch_tbl) local grads,tr_loss,tr_loss_matrix = trainer:compute_gradients_step(input, target) return grads,tr_loss_matrix,bunch_size end -- receives a trainer and the persistent table in read-only mode as last -- argument; returns the validation loss mean and variance local compute_validation_loss = function(trainer, conf) util.omp_set_num_threads(4) local value = "misc/digits.png" local mat = cached(value, make_load_matrix(value), mat_cache) local ds_tbl = cached(value, make_load_dataset(mat,m2), ds_cache) local in_ds = ds_tbl.val_input local out_ds = ds_tbl.val_output local va_loss_mean,va_loss_var = trainer:validate_dataset{ input_dataset = in_ds, output_dataset = out_ds, } util.omp_set_num_threads(1) return va_loss_mean,va_loss_var end -- receives a train_func instance and the persistent table (allows read/write -- operations) local user_finalfn = function(train_func, conf) print(train_func:get_state_string()) util.serialize(train_func, "best_func.lua") end ----------------------------------------------------------------------------- return common.make_map_reduce_task_table { num_reducers = NUM_REDUCERS, dbname = EXP_DBNAME, dbhost = EXP_DBHOST, user_init = user_init, user_taskfn = user_taskfn, user_finalfn = user_finalfn, generate_new_trainer_and_train_func = generate_new_trainer_and_train_func, compute_gradients_and_loss = compute_gradients_and_loss, compute_validation_loss = compute_validation_loss, -- smooth_gradients = true, -- by default it is true } lua-mapreduce-0.4.0/mapreduce/examples/APRIL-ANN/server.lua000066400000000000000000000010271263627443400233270ustar00rootroot00000000000000local mapreduce = require "mapreduce" local connection_string = "localhost" local dbname = "mr_exp_digits" local s = mapreduce.server.new(connection_string, dbname) s:configure{ taskfn = "mapreduce.examples.APRIL-ANN", mapfn = "mapreduce.examples.APRIL-ANN", partitionfn = "mapreduce.examples.APRIL-ANN", reducefn = "mapreduce.examples.APRIL-ANN", finalfn = "mapreduce.examples.APRIL-ANN", init_args = arg, storage = "gridfs", } mapreduce.utils.sleep(4) s:loop() lua-mapreduce-0.4.0/mapreduce/examples/APRIL-ANN/worker.lua000066400000000000000000000002761263627443400233370ustar00rootroot00000000000000local mapreduce = require "mapreduce" local connection_string = "localhost" local dbname = "mr_exp_digits" local w = mapreduce.worker.new(connection_string, dbname) w:execute() lua-mapreduce-0.4.0/mapreduce/examples/WordCount/000077500000000000000000000000001263627443400217615ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/examples/WordCount/finalfn.lua000066400000000000000000000003331263627443400241000ustar00rootroot00000000000000return { init = function() end, finalfn = function(pairs_iterator) for key,values in pairs_iterator do print(values[1],key) end return true -- indicates to remove mongo gridfs result files end } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/init.lua000066400000000000000000000027561263627443400234410ustar00rootroot00000000000000-- string hash function: http://isthe.com/chongo/tech/comp/fnv/ local NUM_REDUCERS = 15 local FNV_prime = 16777619 local offset_basis = 2166136261 local MAX = 2^32 -- arg is for configuration purposes, it will be executed with init_args given -- to the server local init = function(arg) end local taskfn = function(emit) emit(1,"mapreduce/server.lua") emit(2,"mapreduce/worker.lua") emit(3,"mapreduce/test.lua") emit(4,"mapreduce/utils.lua") end local mapfn = function(key,value,emit) for line in io.lines(value) do for w in line:gmatch("[^%s]+") do emit(w,1) end end end local partitionfn = function(key) -- compute hash local h = offset_basis for i=1,#key do h = (h * FNV_prime) % MAX h = bit32.bxor(h, key:byte(i)) end return h % NUM_REDUCERS end local reducefn = function(key,values,emit) local count=0 for _,v in ipairs(values) do count = count + v end emit(count) end local combinerfn = reducefn local finalfn = function(pairs_iterator) for key,values in pairs_iterator do print(values[1],key) end return true -- indicates to remove mongo gridfs result files end return { init = init, taskfn = taskfn, mapfn = mapfn, partitionfn = partitionfn, reducefn = reducefn, combinerfn = combinerfn, finalfn = finalfn, -- This three properties are true for this reduce function. -- Combiners always must to fulfill this properties. associative_reducer = true, commutative_reducer = true, idempotent_reducer = true, } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/mapfn.lua000066400000000000000000000002741263627443400235700ustar00rootroot00000000000000return { init = function() end, mapfn = function(key,value,emit) for line in io.lines(value) do for w in line:gmatch("[^%s]+") do emit(w,1) end end end } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/partitionfn.lua000066400000000000000000000006441263627443400250250ustar00rootroot00000000000000-- string hash function: http://isthe.com/chongo/tech/comp/fnv/ local NUM_REDUCERS = 15 local FNV_prime = 16777619 local offset_basis = 2166136261 local MAX = 2^32 return { init = function() end, partitionfn = function(key) -- compute hash local h = offset_basis for i=1,#key do h = (h * FNV_prime) % MAX h = bit32.bxor(h, key:byte(i)) end return h % NUM_REDUCERS end } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/reducefn.lua000066400000000000000000000006451263627443400242640ustar00rootroot00000000000000local reducefn = function(key,values,emit) local count=0 for _,v in ipairs(values) do count = count + v end emit(count) end return { init = function() end, reducefn = reducefn, combinerfn = reducefn, -- This three properties are true for this reduce function. -- Combiners always must to fulfill this properties. associative_reducer = true, commutative_reducer = true, idempotent_reducer = true, } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/reducefn2.lua000066400000000000000000000003261263627443400243420ustar00rootroot00000000000000local reducefn = function(key,values,emit) local count=0 for _,v in ipairs(values) do count = count + v end emit(count) end return { init = function() end, reducefn = reducefn, combinerfn = reducefn, } lua-mapreduce-0.4.0/mapreduce/examples/WordCount/taskfn.lua000066400000000000000000000005631263627443400237560ustar00rootroot00000000000000-- arg is for configuration purposes, it is allowed in any of the scripts local init = function(arg) -- do whatever you need for initialization parametrized by arg table end return { init = init, taskfn = function(emit) emit(1,"mapreduce/server.lua") emit(2,"mapreduce/worker.lua") emit(3,"mapreduce/test.lua") emit(4,"mapreduce/utils.lua") end } lua-mapreduce-0.4.0/mapreduce/examples/WordCountBig/000077500000000000000000000000001263627443400224035ustar00rootroot00000000000000lua-mapreduce-0.4.0/mapreduce/examples/WordCountBig/taskfn.lua000066400000000000000000000005171263627443400243770ustar00rootroot00000000000000return { -- init is for configuration purposes, it is allowed in any of the scripts init = function(arg) end, taskfn = function(emit) local f = io.popen("ls /home/experimentos/CORPORA/EUROPARL/en-splits/*","r") local i=0 for filename in f:lines() do i=i+1 emit(i,filename) end f:close() end } lua-mapreduce-0.4.0/mapreduce/fs.lua000066400000000000000000000201071263627443400173320ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- This module contains class abstraction which implement gridfs API, allowing -- to use different kinds of intermediate storage, the gridfs storage, the -- shared storage which uses directories in a NFS/samba storage, or the sshfs -- storage which allows to use local file system for writing but uses scp for -- reading. The router function allow to decide which kind of storage will be -- used, and returns an instance of GridFS or sshfs or sharedfs classes. -- module fs local fs = { _VERSION = "0.1", _NAME = "mapreduce.fs", } local utils = require "mapreduce.utils" local make_wildcard_from_mongo_match = function(match_tbl) return match_tbl.filename["$regex"]: gsub("\\%.","."):gsub("%.%*","*"):gsub("[$^]","") end ------------------------------------------------------------------------------ local cursor = {} function cursor:next() local line = self.f:read("*l") if line then return { filename = line } end end function cursor:results() return function() return self:next() end end function cursor:has_more() error("NOT IMPLEMENTED") end function cursor:itcount() error("NOT IMPLEMENTED") end function cursor:is_dead() return false end function cursor:is_tailable() error("NOT IMPLEMENTED") end function cursor:has_result_flag() return false end function cursor:get_id() error("NOT IMPLEMENTED") end function cursor_gc(self) self.f:close() end function cursor:__call(match_tbl) local wildcard_str = make_wildcard_from_mongo_match(match_tbl) local obj = { f = io.popen(string.format("ls -d %s", wildcard_str), "r") } setmetatable(obj, { __index=self, __gc=cursor_gc }) return obj end setmetatable(cursor,cursor) ------------------------------------------------------------------------------ local file_builder = {} local function file_builder_open(self) self.f = assert(io.open(self.tmpname,"w"), string.format("Impossible to open %s", self.tmpname)) self.f:setvbuf("full") end function file_builder:append(data) self.f:write(data) return true end file_builder.write = file_builder.append function file_builder:build(path) local basename = path:match("(.*)/[^/]+") assert(basename and #basename > 0, string.format("Given an incorrect path '%s'", path)) os.execute(string.format("mkdir -p %s 2> /dev/null", basename)) self.f:close() assert( utils.rename(self.tmpname, path) ) file_builder_open(self) return true end function file_builder_gc(self) utils.remove(self.tmpname) end function file_builder:__call() local obj = { tmpname = os.tmpname() } setmetatable(obj, { __index=self, __gc=file_builder_gc }) file_builder_open(obj) return obj end setmetatable(file_builder,file_builder) ------------------------------------------------------------------------------ local sharedfs = {} function sharedfs:list(match_tbl) local match_tbl = match_tbl or { filename = { ["$regex"] = string.format("%s/.*", self.path) } } return cursor(match_tbl) end function sharedfs:remove_file(filename) return utils.remove(filename) end function sharedfs:__call(path,hostnames) local obj = { path=path } os.execute(string.format("mkdir -p %s 2> /dev/null", path)) setmetatable(obj, { __index=self }) return obj end setmetatable(sharedfs,sharedfs) ------------------------------------------------------------------------------ local sshfs = {} function sshfs:list(match_tbl) local match_tbl = match_tbl or { filename = { ["$regex"] = string.format("%s/.*", self.path) } } local wildcard_str = make_wildcard_from_mongo_match(match_tbl) local processed = {} for _,hostname in ipairs(self.hostnames) do if not processed[hostname] then local ok,exit,signal = os.execute(string.format("scp -CB %s:%s %s/", hostname, wildcard_str, self.tmpname)) assert(ok, string.format("Impossible to SCP remote files from %s:%s", hostname, wildcard_str)) processed[hostname] = true end end match_tbl.filename["$regex"] = match_tbl.filename["$regex"]:gsub(self.path,self.tmpname) return cursor(match_tbl) end function sshfs:remove_file(filename) local filename = filename:gsub(self.path,self.tmpname) return utils.remove(filename) end function sshfs_gc(self) -- remove only if it is empty utils.remove(self.tmpname) end function sshfs:__call(path,hostnames) local obj = { path=path:gsub("/$",""), tmpname=os.tmpname(), hostnames=hostnames } utils.remove(obj.tmpname) os.execute(string.format("mkdir -p %s 2> /dev/null", obj.tmpname)) setmetatable(obj, { __index=self, __gc=sshfs_gc }) return obj end setmetatable(sshfs, sshfs) ------------------------------------------------------------------------------ local router = function(cnn, hostnames, storage, path) if storage == "gridfs" then return cnn:gridfs(), function() return cnn:grid_file_builder() end, function(filename) return utils.gridfs_lines_iterator(cnn:gridfs(), filename) end elseif storage == "sshfs" and hostnames and #hostnames>0 then local obj = sshfs(path,hostnames) return obj, function() return file_builder() end, function(filename) local filename = filename:gsub(path,obj.tmpname) return io.lines(filename) end elseif ( storage == "shared" or (storage == "sshfs" and (not hostnames or #hostnames == 0)) ) then return sharedfs(path), function() return file_builder() end, function(filename) return io.lines(filename) end else error(string.format("Given incorrect storage %s", storage)) end end ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- fs.utest = function() assert(make_wildcard_from_mongo_match({ filename = { ["$regex"] = "/tmp/.*" } }) == "/tmp/*") local cnn = require "mapreduce.cnn" local path = os.tmpname() utils.remove(path) os.execute("mkdir -p " .. path) local c = cnn("localhost", "test") local db = c:connect() db:drop_collection("test.fs.files") db:drop_collection("test.fs.chunks") local hostnames = { "localhost" } local lines = "first line\nsecond line\n" for _,storage in ipairs{ "gridfs", "shared", "sshfs" } do local fs,make_builder,make_lines_iterator = router(c, hostnames, storage, path) local builder = make_builder() local inv_files = {} for _,file in ipairs{ "test1", "test2" } do inv_files[file] = true local file = (path .. "/" .. file):gsub("//", "/") builder:append(lines) builder:build(file) end local q = fs:list() for v in q:results() do assert(inv_files[v.filename:match("([^/]+)$")]) local lines_tbl = {} for line in make_lines_iterator(v.filename) do table.insert(lines_tbl, line) end assert(table.concat(lines_tbl,"\n").."\n" == lines) assert( fs:remove_file(v.filename) ) end end end ------------------------------------------------------------------------------ fs.router = router fs.sshfs = sshfs fs.sharedfs = sharedfs return fs lua-mapreduce-0.4.0/mapreduce/heap.lua000066400000000000000000000053541263627443400176460ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] local heap = { _VERSION = "0.1", _NAME = "mapreduce.heap", } local function parent(p) return math.floor(p/2) end local function left(p) return 2*p end local function right(p) return 2*p+1 end function heap:top() return self.data[1] end function heap:pop() local data, cmp = self.data, self.cmp local v = data[#data] data[#data] = nil if #data == 0 then return end -- heapify(1) local pos = 1 while true do local l,r = left(pos), right(pos) local lv, rv = data[l], data[r] if not lv then break end local child = (rv and cmp(rv,lv) and r) or l if cmp(data[child],v) then data[pos] = data[child] pos = child else break end end data[pos] = v end function heap:push(v) local data, cmp = self.data, self.cmp local pos = #data + 1 -- bubble up while pos > 1 do local p = parent(pos) local pv = data[p] if cmp(v,pv) then data[pos] = data[p] pos = p else break end end data[pos] = v end function heap:clear() self.data = {} end function heap:size() return #self.data end function heap:empty() return #self.data == 0 end function heap:__call(cmp) local obj = { cmp = cmp or function(a,b) return a < b end, data = {}, } setmetatable(obj, { __index = self, __len = function(self) return #self.data end }) return obj end setmetatable(heap, heap) ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- heap.utest = function() local t = { 20, 10, 15, 1 } local h = heap() assert(h:empty()) assert(h:top() == nil) assert(h:size() == 0) h:push(30) assert(h:top() == 30) h:clear() assert(h:empty()) assert(h:top() == nil) assert(h:size() == 0) for i=1,#t do h:push(t[i]) end table.sort(t) for i=1,#t do assert(h:top() == t[i]) h:pop() end assert(h:top() == nil) end return heap lua-mapreduce-0.4.0/mapreduce/init.lua000066400000000000000000000024231263627443400176660ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] local worker = require "mapreduce.worker" local server = require "mapreduce.server" local utils = require "mapreduce.utils" local tuple = require "mapreduce.tuple" local persistent_table = require "mapreduce.persistent_table" local mapreduce = { _VERSION = "0.4.0", _NAME = "mapreduce", worker = worker, server = server, utils = utils, tuple = tuple, persistent_table = persistent_table, } -- integrity test mapreduce.utest = function(connection_string, dbname) server.utest(connection_string, dbname) end return mapreduce lua-mapreduce-0.4.0/mapreduce/job.lua000066400000000000000000000330001263627443400174700ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- The job class is used by workers to execute map/reduce job. This class allows -- to write job status and to update job statistics in MongoDB. Execution of -- user map/reduce/combiner modules is done in job class. Intermediate data is -- written here in the storage given at 'task' collection. local job = { _VERSION = "0.4", _NAME = "job", } local utils = require "mapreduce.utils" local fs = require "mapreduce.fs" local tuple = require "mapreduce.tuple" local STATUS = utils.STATUS local TASK_STATUS = utils.TASK_STATUS local serialize_table_ipairs = utils.serialize_table_ipairs local merge_iterator = utils.merge_iterator local keys_sorted = utils.keys_sorted local escape = utils.escape -- local cache = {} local function cached(func) cache[func] = cache[func] or {} local local_cache = cache[func] return function(key) if not local_cache[key] then local result = func(key) local_cache[key] = result return result else return local_cache[key] end end end local function reset_cache() cache = {} end -- PRIVATE FUNCTIONS AND METHODS -- loads the required Lua module, sets the "emit" function, executes init -- function if needed, and returns the resulting function local initialized = {} local funcs = { } local function job_get_func(self, fname, func, args) local f = funcs[fname] if not f then f = { m = require(fname) } if f.m.init and not initialized[f.m.init] then f.m.init(args) initialized[f.m.init] = true end assert(f.m[func]) -- sanity check funcs[fname] = f end if func == "mapfn" then local MAX_MAP_RESULT = utils.MAX_MAP_RESULT local result = {} local copy_table_ipairs = utils.copy_table_ipairs local clear_table = utils.clear_table self.result = result self.emit = function(key,value) local key,value = tuple(key),tuple(value) -- assert(tostring(key), -- "emit function must receive a convertible to string key") local result = result result[key] = result[key] or {} local N = #result[key] -- faster than table.insert result[key][ N+1 ] = value if self.combiner and N > MAX_MAP_RESULT then self.combiner(key, result[key],self.combiner_emit) copy_table_ipairs(result[key], self.combiner_result) clear_table(self.combiner_result) end end elseif func == "reducefn" then self.result = {} self.emit = function(value) -- faster than table.insert self.result[ #self.result+1 ] = tuple(value) end self.associative_reducer = f.m.associative_reducer self.commutative_reducer = f.m.commutative_reducer self.idempotent_reducer = f.m.idempotent_reducer elseif func == "combinerfn" then self.combiner_result = {} self.combiner_emit = function(value) -- faster than table.insert self.combiner_result[ #self.combiner_result+1 ] = tuple(value) end end return f.m[func] end local function job_mark_as_finished(self) assert(self.job_tbl) local db = self.cnn:connect() assert( db:update(self.jobs_ns, { _id = self:get_id(), }, { ["$set"] = { status = STATUS.FINISHED, finished_time = utils.time(), }, }, false, false) ) end function job_mark_as_written(self,cpu_time) self.written = true assert(self.job_tbl) local db = self.cnn:connect() assert( db:update(self.jobs_ns, { _id = self:get_id(), }, { ["$set"] = { status = STATUS.WRITTEN, written_time = utils.time(), cpu_time = cpu_time, real_time = utils.time() - self.t, }, }, false, false) ) end function job_prepare_map(self, g, combiner_fname, partitioner_fname, init_args, storage, path) local partitioner = cached( job_get_func(self, partitioner_fname, "partitionfn", init_args) ) -- combiner, apply the reduce function before put result to database local combiner if combiner_fname then combiner = job_get_func(self, combiner_fname, "combinerfn", init_args) end self.combiner = combiner local map_key,map_value = self:get_pair() -- this closure is the responsible for all the map job return function() -- take local variables local assert = assert local tonumber = tonumber local combiner = combiner local partitioner = partitioner local string_format = string.format local escape = escape local math_floor = math.floor local serialize_table_ipairs = serialize_table_ipairs local clock1 = os.clock() local combiner_emit = self.combiner_emit local copy_table_ipairs = utils.copy_table_ipairs local clear_table = utils.clear_table -------------------------------------------------------------------------- -------------------------------------------------------------------------- g(map_key,map_value,self.emit) -- executes the MAP function, the result is -------------------------------------------------------------------------- -------------------------------------------------------------------------- -- self.result the job is marked as finished, but not written job_mark_as_finished(self) -- local results_ns = self.results_ns -- partition of the map result using partition function; additionally, the -- data is stored sorted by key; data is appended to builder local result = self.result or {} local fs,make_builder,make_lines_iterator = fs.router(self.cnn,nil, storage,path) local keys = keys_sorted(result) local builders = {} for _,key in ipairs(keys) do local values = result[key] if #values > 1 and combiner then combiner(key,values,combiner_emit) copy_table_ipairs(values, self.combiner_result) clear_table(self.combiner_result) end local part_key = partitioner(key) local part_key = assert(tonumber(part_key), "Partition key must be a number") assert(math_floor(part_key) == part_key, "Partition key must be an integer") local result_ns = string_format("%s.P%d.M%s", results_ns, part_key, map_key) local builder = builders[result_ns] or make_builder() builders[result_ns] = builder local key_str = escape(key) local values_str = serialize_table_ipairs(values) builder:append(string.format("return %s,%s\n", key_str, values_str)) end -- create all the files for result_ns,builder in pairs(builders) do local fs_filename = string.format("%s/%s",path,result_ns) fs:remove_file(fs_filename) assert( builder:build(fs_filename) ) end local clock2 = os.clock() local elapsed_time = clock2 - clock1 -- job is marked as written to the database job_mark_as_written(self, elapsed_time) return elapsed_time end end function job_prepare_reduce(self, g, storage, path) local key,value = self:get_pair() return function() -- take local variables local copy_table_ipairs = utils.copy_table_ipairs local clear_table = utils.clear_table local associative = self.associative_reducer local commutative = self.commutative_reducer local idempotent = self.idempotent_reducer local assert = assert local string_format = string.format local escape = escape local clock1 = os.clock() -- in reduce jobs, the value is a reference with the basename of the fs -- filenames related to the given reduce job local part_key = key local job_file = value.file local res_file = value.result local mappers = value.mappers local gridfs = self.cnn:gridfs() local builder = self.cnn:grid_file_builder() gridfs:remove_file(res_file) -- take all the files which match the given job_file name local fs,make_builder,make_lines_iterator = fs.router(self.cnn,mappers, storage,path) local filenames = {} local match_str = string.format("^%s\\..*", job_file) local list = fs:list({ filename = { ["$regex"] = match_str } }) for v in list:results() do table.insert(filenames, v.filename) end -------------------------------------------------------------------------- -------------------------------------------------------------------------- -- iterate over a merge of all the input filenames if associative and commutative and idempotent then for k,v in merge_iterator(fs, filenames, make_lines_iterator) do if #v > 1 then g(k,v,self.emit) -- executes the REDUCE function copy_table_ipairs(v, self.result) clear_table(self.result) end -- write the result to mongo builder:append(string_format("return %s,%s\n", escape(k), serialize_table_ipairs(v))) end else for k,v in merge_iterator(fs, filenames, make_lines_iterator) do g(k,v,self.emit) -- executes the REDUCE function copy_table_ipairs(v, self.result) clear_table(self.result) -- write the result to mongo builder:append(string_format("return %s,%s\n", escape(k), serialize_table_ipairs(v))) end end -------------------------------------------------------------------------- -------------------------------------------------------------------------- assert( builder:build(res_file) ) local clock2 = os.clock() local elapsed_time = clock2-clock1 -- job is marked as as written directly job_mark_as_written(self, elapsed_time) -- remove all map result fs files for _,name in ipairs(filenames) do fs:remove_file(name) end return elapsed_time end end -- PUBLIC METHODS function job:execute() return self.fn() end function job:get_id() assert(self.job_tbl) return self.job_tbl._id end function job:get_pair() assert(self.job_tbl) return self.job_tbl._id,self.job_tbl.value end function job:status_string() return self:get_id() end function job:get_results_ns() return self.results_ns end function job:mark_as_broken() if not self.written then assert(self.job_tbl) local db = self.cnn:connect() assert( db:update(self.jobs_ns, { _id = self:get_id(), }, { ["$set"] = { status = STATUS.BROKEN, broken_time = utils.time(), }, ["$inc"] = { repetitions = 1, }, }, false, false) ) end end -- constructor, receives a connection and a task instance function job:__call(cnn, job_tbl, task_status, fname, init_args, jobs_ns, results_ns, not_executable, combiner, partitioner, storage, path) local obj = { cnn = cnn, job_tbl = job_tbl, jobs_ns = jobs_ns, results_ns = results_ns, t = utils.time(), } setmetatable(obj, { __index=self }) -- local fn,g,func if task_status == TASK_STATUS.MAP then func = "mapfn" elseif task_status == TASK_STATUS.REDUCE then func = "reducefn" else error("Incorrect task_status: " .. tostring(task_status) ) end if not not_executable then g = job_get_func(obj, fname, func, args, task_status == TASK_STATUS.MAP) end if not not_executable then if task_status == TASK_STATUS.MAP then fn = job_prepare_map(obj, g, combiner, partitioner, init_args, storage, path) elseif task_status == TASK_STATUS.REDUCE then fn = job_prepare_reduce(obj, g, storage, path) end end obj.fn = fn or function() error("Forbidden execution of jobs here") end return obj end setmetatable(job,job) ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- job.utest = function() local f = cached(function(i) return i end) for j=1,2 do for i=1,10 do assert(f(i) == i) end end end ------------------------------------------------------------------------ job.reset_cache = reset_cache return job lua-mapreduce-0.4.0/mapreduce/persistent_table.lua000066400000000000000000000205621263627443400222760ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- The class persistent_table allows to store data in a collection of -- singletons, allowing to retrieve data if after a system breakdown, or to use -- several data configuration through several distributed processes. local persistent_table = { _VERSION = "0.2", _NAME = "mapreduce.persistent_table", } local mongo = require "mongo" local utils = require "mapreduce.utils" local cnn = require "mapreduce.cnn" local LOCK_SLEEP = 0.1 -- in seconds -- PUBLIC METHODS local methods = {} -- updates current object with MongoDB collection; in case of inconsistency -- between this object and database, an assert will be thrown function methods:update() local self = getmetatable(self).obj local cnn = self.cnn local dirty = self.dirty local content = self.content local db = cnn:connect() local query = { _id = content._id, } local update = { } if dirty then query.timestamp = content.timestamp content.timestamp = nil local t = {} update["$set"] = t for k,v in pairs(content) do t[k] = v end t._id = nil update["$inc"] = { timestamp = 1 } else update["$set"] = { __dummy__ = true } end -- findAndModify returns the document AFTER the modification (new=true) local result = assert( db:run_command(self.dbname, { cmd = "findAndModify", findAndModify = self.document, query = query, update = update, upsert = false, new = true, }) ) local remote_content = result.value assert(mongo.type(remote_content) ~= "mongo.NULL", "Impossible to update, data is not consistent") self.dirty = false end -- removes the collection at the database, allowing to start from zero function methods:drop() local aux = self local self = getmetatable(self).obj local db = self.cnn:connect() -- findAndModify returns the document AFTER the modification (new=true) local result = assert( db:run_command(self.dbname, { cmd = "findAndModify", findAndModify = self.document, query = { _id = self.content._id }, update = { __dummy__ = true }, upsert=true, new=true, }) ) assert(mongo.type(result.value) ~= "mongo.NULL") self.content = result.value end local reserved = { _id=true, timestamp=true, set=true, update=true, drop=true, read_only=true, dirty=true, locked=true, __dummy__=true } -- sets a collection of pairs key,value from the given table function methods:set(tbl) local self = getmetatable(self).obj assert(not self.read_only, "Unable to write in a read_only persistent table") for key,value in pairs(tbl) do utils.assert_check(value) if reserved[key] then error(string.format("%s field is reserved",key)) end rawset(self.content,key,value) end self.dirty=true end local local_set = methods.set function methods:lock() local self = getmetatable(self).obj local db = self.cnn:connect() local content = self.content local remote_content repeat -- findAndModify returns the document value BEFORE the modification -- (new=false) ... local result = assert( db:run_command(self.dbname, { cmd = "findAndModify", findAndModify = self.document, query = { _id = content._id }, update = { ["$set"] = { locked = true }, }, upsert = false, new = false, }) ) remote_content = result.value assert(mongo.type(remote_content) ~= "mongo.NULL") -- and before the modification it must be locked=false, otherwise sleep. if remote_content.locked then utils.sleep(LOCK_SLEEP) end until not remote_content.locked self.locked = true end function methods:unlock() local self = getmetatable(self).obj local db = self.cnn:connect() local content = self.content local remote_content -- findAndModify returns the document value BEFORE the modification -- (new=false) ... local result = assert( db:run_command(self.dbname, { cmd = "findAndModify", findAndModify = self.document, query = { _id = content._id }, update = { ["$set"] = { locked = false }, }, upsert = false, new = false, }) ) remote_content = result.value assert(mongo.type(remote_content) ~= "mongo.NULL") self.locked = false end function methods:read_only(v) local self = getmetatable(self).obj self.read_only = v end function methods:dirty() local self = getmetatable(self).obj return self.dirty end ------------------------------------------------------------------------------ -- constructor using encapsulation design pattern in Lua function persistent_table:__call(name, cnn_string, dbname, document, auth_table) assert(type(name) == "string","First argument is a string name for the table") local cnn_string = cnn_string or "localhost" local dbname = dbname or "tmp" local document = document or "singletons" -- obj is hidden inside a closure local obj = { cnn = cnn(cnn_string, dbname, auth_table), name = name, dirty = false, read_only = false, dbname = dbname, document = document, singleton_ns = dbname .. "." .. document, content = { _id = name }, } local db = obj.cnn:connect() -- findAndModify returns the document AFTER the modification (new=true) local result = assert( db:run_command(obj.dbname, { cmd = "findAndModify", findAndModify = obj.document, query = obj.content, update = { ["$set"] = { __dummy__ = true }, }, upsert=true, new=true, }) ) assert(mongo.type(result.value) ~= "mongo.NULL") obj.content = result.value -- visible_table has been prepared to capture obj table as closure of its -- metatable, allowing to implement data encapsulation design pattern in Lua local visible_table = {} setmetatable(visible_table, { -- retrieve the obj table from an instance of visible_table obj = obj, -- unlock for just in case __gc = function() if obj.locked == true then self:unlock() end end, -- index a key __index = function(self,key) if methods[key] then return methods[key] else return obj.content[key] end end, -- sets the value of a key,value pair __newindex = function(self,key,value) if value == nil then obj.content[key] = nil obj.dirty = true else local_set(self,{ [key]=value }) end end, -- shows a JSON string __tostring = function(self) -- copy to aux all the content fields except the reserved -- fields local aux = {} for k,v in pairs(obj.content) do if not reserved[k] then aux[k] = v end end return utils.tojson(aux) end, }) visible_table:update() return visible_table end setmetatable(persistent_table, persistent_table) ------------------------------------------------------------------------------ persistent_table.utest = function() local conf = persistent_table("conf") conf:drop() conf:set({ key = "test" }) conf:update() local conf2 = persistent_table("conf") conf2:update() assert(conf.key == conf2.key) end ------------------------------------------------------------------------------ return persistent_table lua-mapreduce-0.4.0/mapreduce/server.lua000066400000000000000000000607671263627443400202500ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- The server is one of the most important classes, together with job -- class. Server allows to configure a (iterative) MapReduce task. It configures -- 'map_jobs' and 'red_jobs' collections, where workers look to run jobs. The -- programmer needs to execute server:configure(...) method in order to setup -- the MapReduce task you want to execute. server:loop() method executes the -- task and take note of statistics which will be written using a task object. --[[ Example of map_jobs collection for wordcount > db.map_jobs.find().limit(2).pretty() { "_id" : "1", "value" : "mapreduce/server.lua", "creation_time" : 1400225797.659415, "started_time" : 1400225798.321171, "finished_time" : 1400225798.328765, "written_time" : 1400225798.346263, "cpu_time" : 0.017617999999999998, "real_time" : 0.022084951400756836, "status" : 4, "tmpname" : "lua_QcozTk", "worker" : "HOSTNAME" } { "_id" : "2", "value" : "mapreduce/worker.lua", "creation_time" : 1400225797.659444, "started_time" : 1400225798.346911, "finished_time" : 1400225798.350307, "written_time" : 1400225798.358262, "cpu_time" : 0.005623000000000003, "real_time" : 0.008959054946899414, "status" : 4, "tmpname" : "lua_QcozTk", "worker" : "HOSTNAME" } ]] --[[ Example of red_jobs collection for wordcount > db.red_jobs.find().limit(2).pretty() { "_id" : "1", "value" : { "result" : "result.P1", "mappers" : [ "HOSTNAME", "HOSTNAME", "HOSTNAME", "HOSTNAME" ], "file" : "/tmp/lua_sQHr2G/map_results.P1" }, "creation_time" : 1400225798.665791, "started_time" : 1400225799.41176, "written_time" : 1400225799.423528, "cpu_time" : 0.006439999999999994, "real_time" : 0.008752822875976562, "status" : 4, "tmpname" : "lua_QcozTk", "worker" : "HOSTNAME", } { "_id" : "2", "value" : { "result" : "result.P2", "mappers" : [ "HOSTNAME", "HOSTNAME", "HOSTNAME", "HOSTNAME" ], "file" : "/tmp/lua_sQHr2G/map_results.P2" }, "creation_time" : 1400225798.665803, "started_time" : 1400225799.426011, "written_time" : 1400225799.435997, "cpu_time" : 0.004669999999999994, "real_time" : 0.006902933120727539, "status" : 4, "tmpname" : "lua_QcozTk", "worker" : "HOSTNAME" } ]] local server = { _VERSION = "0.3", _NAME = "mapreduce.server", } local utils = require "mapreduce.utils" local task = require "mapreduce.task" local cnn = require "mapreduce.cnn" local fs = require "mapreduce.fs" local tuple = require "mapreduce.tuple" local MAX_JOB_RETRIES = utils.MAX_JOB_RETRIES local DEFAULT_HOSTNAME = utils.DEFAULT_HOSTNAME local DEFAULT_IP = utils.DEFAULT_IP local DEFAULT_DATE = utils.DEFAULT_DATE local STATUS = utils.STATUS local TASK_STATUS = utils.TASK_STATUS local escape = utils.escape local serialize_table_ipairs = utils.serialize_table_ipairs local make_job = utils.make_job local gridfs_lines_iterator = utils.gridfs_lines_iterator local get_storage_from = utils.get_storage_from -- PRIVATE FUNCTIONS AND METHODS local function keys_of(t) local out = {} for k,_ in pairs(t) do table.insert(out, k) end return out end local function count_digits(n) -- sanity check assert(n >= 0, "Only valid for positive integers") if n == 0 then return 1 end local c = 0 while n > 0 do n = math.floor(n/10) c = c + 1 end return c end local function wrap_pcall(func, ...) local ok,result = xpcall(func, debug.traceback, ...) if not ok then -- io.stderr:write(string.format("%s\n", result)) result = 0 end return result end local function compute_real_time(db, ns) local out_min = assert( db:mapreduce(ns, [[ function() { emit(0, ( this.started_time ) ? this.started_time : this.creation_time ) } ]], [[ function(k,v) { var min=v[0]; for (var i=1; imax) max=v[i]; return max; }]]) ) return out_max.results[1].value - out_min.results[1].value end local function compute_sum(db, ns, field) local result = assert( db:mapreduce(ns, string.format([[ function() { emit(0, this.%s) } ]], field), [[ function(k,v) { return Array.sum(v); }]]) ) return result.results[1].value end -- returns a coroutine.wrap which returns true until all tasks are finished local function make_task_coroutine_wrap(self,ns) local db = self.cnn:connect() local N = db:count(ns) return coroutine.wrap(function() repeat local db = self.cnn:connect() -- remove broken jobs with >= MAX_JOB_RETRIES -- FIXME: check the write concern assert( db:update(ns, { status = STATUS.BROKEN, repetitions = { ["$gte"] = MAX_JOB_RETRIES }, }, { ["$set"] = { status = STATUS.FAILED }, }, false, -- no create a new document if not exists true) ) -- update all the matches -- count written or failed jobs local M = db:count(ns, { ["$or"] = { { status = STATUS.WRITTEN }, { status = STATUS.FAILED }, } }) if M then io.stderr:write(string.format("\r\t %6.1f %% ", M/N*100)) io.stderr:flush() local ids = {} local q = self.cnn:get_errors() for v in q:results() do table.insert(ids, v._id) io.stderr:write(string.format("\nError from %s: %s\n", v.worker, v.msg)) io.stderr:flush() end if #ids > 0 then self.cnn:remove_errors(ids) end end if not M or M < N then coroutine.yield(true) end until M == N io.stderr:write("\n") end) end -- removes all the tasks which are not WRITTEN and not FAILED local function remove_pending_tasks(db,ns) return db:remove(ns, { ["$or"] = { { status = STATUS.BROKEN, }, { status = STATUS.WAITING }, { status = STATUS.FINISHED }, { status = STATUS.RUNNING }, }, }, false) end -- insert jobs in mongo db and returns a coroutine ready to be executed as an -- iterator local function server_prepare_map(self) local count = 0 local db = self.cnn:connect() local map_jobs_ns = self.task:get_map_jobs_ns() remove_pending_tasks(db, map_jobs_ns) -- create map tasks in mongo database local keys_check = {} self.taskfn.taskfn(function(key,value) count = count + 1 local key = tuple(key) assert(not keys_check[key], string.format("Duplicate key: %s", key)) keys_check[key] = true local tvalue = type(value) if tvalue == "table" then local json_value = utils.tojson(value) assert(#json_value <= utils.MAX_TASKFN_VALUE_SIZE, "Exceeded maximum taskfn value size") end -- FIXME: check how to process task keys which are -- defined by a previously broken execution and didn't -- belong to the current task execution assert( db:insert(map_jobs_ns, make_job(key,value)) ) end) self.task:set_task_status(TASK_STATUS.MAP) -- this coroutine WAITS UNTIL ALL MAPS ARE DONE return make_task_coroutine_wrap(self, map_jobs_ns),count end -- insert the job in the mongo db and returns a coroutine local function server_prepare_reduce(self) local db = self.cnn:connect() local dbname = self.cnn:get_dbname() local map_results_ns = self.task:get_map_results_ns() local red_jobs_ns = self.task:get_red_jobs_ns() remove_pending_tasks(db, red_jobs_ns) -- list all the mappers local map_jobs_ns = self.task:get_map_jobs_ns() local query = db:query(map_jobs_ns,{}) local map_hostnames = {} for q in query:results() do map_hostnames[tostring(q._id)] = q.worker end -- list the filenames generated by mappers in order to create the reduce jobs local storage,path = self.task:get_storage() local fs,make_builder,make_lines_iterator = fs.router(self.cnn,nil, storage,path) local match_str = string.format("^%s/.*P.*M.*$",path):gsub("//","/") local filenames = {} local list = fs:list({ filename = { ["$regex"] = match_str } }) local part_keys = {} local max_part_key = 0 local mappers_by_part_key = {} for obj in list:results() do local filename = obj.filename -- sanity check assert(filename:match(match_str)) -- create reduce jobs in mongo database, from partitioned space local part_key,mapper_key = filename:match("^.*.P([^%.]+)%.M([^%.]*)$") part_key = assert(tonumber(part_key)) part_keys[part_key] = true max_part_key = math.max(max_part_key, part_key) -- annotate the mapper mappers_by_part_key[part_key] = mappers_by_part_key[part_key] or {} mappers_by_part_key[part_key][ map_hostnames[mapper_key] ] = true end local part_key_digits = count_digits(max_part_key) local result_str_format = "%s.P%0" .. tostring(part_key_digits) .. "d" local count=0 for part_key,_ in pairs(part_keys) do count = count + 1 local value = { mappers = keys_of(mappers_by_part_key[part_key]), file = string.format("%s/%s.P%d", path, map_results_ns, part_key), result = string.format(result_str_format, self.result_ns, part_key), } self.cnn:annotate_insert(red_jobs_ns, make_job(part_key, value)) end self.cnn:flush_pending_inserts(0) self.task:set_task_status(TASK_STATUS.REDUCE) -- this coroutine WAITS UNTIL ALL REDUCES ARE DONE return make_task_coroutine_wrap(self, red_jobs_ns),count end local function server_drop_collections(self) local db = self.cnn:connect() local dbname = self.cnn:get_dbname() -- drop all the collections for _,name in ipairs(db:get_collections(dbname)) do if name ~= "system_indexes" then db:drop_collection(dbname .. "." .. name) end end local gridfs = self.cnn:gridfs() local list = gridfs:list() for v in list:results() do gridfs:remove_file(v.filename) end end -- finalizer for the map-reduce process local function server_final(self) -- FIXME: self.result_ns could contain especial characters, it will be -- necessary to escape them local match_str = string.format("^%s",self.result_ns) local gridfs = self.cnn:gridfs() local files = gridfs:list({ filename = { ["$regex"] = match_str } }) local files_tbl local k=0 local current_filename local lines_iterator -- iterator which is given to final function, allows to traverse all the -- results by pairs key,value local pair_iterator = function() if not files_tbl then files_tbl = {} for v in files:results() do assert(v.filename:match(match_str)) table.insert(files_tbl, v.filename) end table.sort(files_tbl) end local line repeat if lines_iterator then line = lines_iterator() end if not line then k=k+1 current_filename = files_tbl[k] if current_filename then lines_iterator = gridfs_lines_iterator(gridfs,current_filename) end end until current_filename == nil or line ~= nil if line then return load(line)() end end -- the reply could be: false/nil, true, "loop" local reply = self.finalfn.finalfn(pair_iterator) local remove_all = (reply == true) or (reply == "loop") if reply ~= "loop" and reply ~= true and reply ~= false and reply ~= nil then io.stderr:write("# WARNING!!! INCORRECT FINAL RETURN: " .. tostring(reply) .. "\n") end -- drop collections, except reduce result and task status local db = self.cnn:connect() -- local task = self.task if reply == "loop" then io.stderr:write("# LOOP again\n") db:drop_collection(task:get_map_jobs_ns()) db:drop_collection(task:get_red_jobs_ns()) else self.finished = true task:set_task_status(TASK_STATUS.FINISHED) end -- remove gridfs files local gridfs = self.cnn:gridfs() local list = gridfs:list() for v in list:results() do if not v.filename:match(match_str) or remove_all then gridfs:remove_file(v.filename) end end end -- SERVER METHODS local server_methods = {} -- configures the server with the script string function server_methods:configure(params) params.storage = string.format("%s:%s",get_storage_from(params.storage,true)) self.configured = true self.configuration_params = params self.init_args = params.init_args local dbname = self.dbname local scripts = {} self.result_ns = params.result_ns or "result" assert(params.taskfn and params.mapfn and params.partitionfn and params.reducefn, "Fields taskfn, mapfn, partitionfn and reducefn are mandatory") for _,name in ipairs{ "taskfn", "mapfn", "partitionfn", "reducefn", "finalfn", "combinerfn" } do assert( (params[name] and type(params[name]) == "string") or (not params[name] and (name=="finalfn" or name == "combinerfn") ), string.format("Needs a %s module with %s function", name, name)) if params[name] then local aux = require(params[name]) assert(type(aux) == "table", string.format("Module %s must return a table", name)) assert(aux[name], string.format("Module %s must return a table with the field %s", name, name)) assert(aux.init, string.format("Init function is needed: %s", name)) scripts[name] = params[name] end end local db = self.cnn:connect() -- self.taskfn = require(scripts.taskfn) if scripts.finalfn then self.finalfn = require(scripts.finalfn) else self.finalfn = { finalfn = function() end, init = function() end } end local init_functions = { [self.taskfn.init] = self.taskfn.init, [self.finalfn.init] = self.finalfn.init, } for _,init in pairs(init_functions) do init(self.init_args) end self.mapfn = params.mapfn self.reducefn = params.reducefn self.partitionfn = params.partitionfn end -- makes all the map-reduce process, looping into the coroutines until all tasks -- are done function server_methods:loop() assert(self.configured, "Call to server:configure(...) method is mandatory") local it = 0 repeat local skip_map,initialize=false,true if it == 0 then -- in the first iteration, we check if the task is a new fresh execution -- or if a previous broken task exists self.task:update() if self.task:has_status() then local status = self.task:get_task_status() if status == TASK_STATUS.REDUCE then -- if the task was in reduce state, skip map jobs and re-run reduce io.stderr:write("# WARNING: TRYING TO RESTORE A BROKEN TASK\n") skip_map = true initialize = false self.configuration_params.storage = string.format("%s:%s", self.task:get_storage()) elseif status == TASK_STATUS.FINISHED then -- if the task was finished, therefore it is a shit, drop old data server_drop_collections(self) else -- otherwise, the task is in WAIT or MAP states, try to restore from -- there initialize = false end end -- if task has status end -- if it == 0 if initialize then -- count one iteration it = it+1 -- create task object self.task:create_collection(TASK_STATUS.WAIT, self.configuration_params, it) else it = self.task:get_iteration() self.task:create_collection(self.task:get_task_status(), self.configuration_params, it) end io.stderr:write(string.format("# Iteration %d\n", it)) local start_time = utils.time() self.task:insert_started_time(start_time) if not skip_map then -- MAP EXECUTION io.stderr:write("# \t Preparing Map\n") local do_map_step,map_count = server_prepare_map(self) collectgarbage("collect") io.stderr:write(string.format("# \t Map execution, size= %d\n", map_count)) while do_map_step() do utils.sleep(utils.DEFAULT_SLEEP) collectgarbage("collect") end end local db = self.cnn:connect() local map_count = db:count(self.task:get_map_jobs_ns()) -- REDUCE EXECUTION collectgarbage("collect") io.stderr:write("# \t Preparing Reduce\n") local do_reduce_step = server_prepare_reduce(self) local db = self.cnn:connect() local red_count = db:count(self.task:get_red_jobs_ns()) collectgarbage("collect") io.stderr:write(string.format("# \t Reduce execution, num_files= %d size= %d\n", red_count * map_count, red_count)) while do_reduce_step() do utils.sleep(utils.DEFAULT_SLEEP) collectgarbage("collect") end -- TIME local end_time = utils.time() local total_time = end_time - start_time self.task:insert_finished_time(end_time) -- STATISTICS local map_sum_cpu_time = wrap_pcall(compute_sum, db, self.task:get_map_jobs_ns(), "cpu_time") local red_sum_cpu_time = wrap_pcall(compute_sum, db, self.task:get_red_jobs_ns(), "cpu_time") local map_sum_real_time = wrap_pcall(compute_sum, db, self.task:get_map_jobs_ns(), "real_time") local red_sum_real_time = wrap_pcall(compute_sum, db, self.task:get_red_jobs_ns(), "real_time") local map_real_time = wrap_pcall(compute_real_time, db, self.task:get_map_jobs_ns()) local red_real_time = wrap_pcall(compute_real_time, db, self.task:get_red_jobs_ns()) io.stderr:write(string.format("# Map sum(cpu_time) %f\n", map_sum_cpu_time)) io.stderr:write(string.format("# Reduce sum(cpu_time) %f\n", red_sum_cpu_time)) io.stderr:write(string.format("# Sum(cpu_time) %f\n", map_sum_cpu_time + red_sum_cpu_time)) io.stderr:write(string.format("# Map sum(real_time) %f\n", map_sum_real_time)) io.stderr:write(string.format("# Reduce sum(real_time) %f\n", red_sum_real_time)) io.stderr:write(string.format("# Sum(real_time) %f\n", map_sum_real_time + red_sum_real_time)) io.stderr:write(string.format("# Sum(sys_time) %f\n", map_sum_real_time + red_sum_real_time - map_sum_cpu_time - red_sum_cpu_time)) io.stderr:write(string.format("# Map cluster time %f\n", map_real_time)) io.stderr:write(string.format("# Reduce cluster time %f\n", red_real_time)) io.stderr:write(string.format("# Cluster time %f\n", map_real_time + red_real_time)) -- local num_failed_maps = db:count(self.task:get_map_jobs_ns(), { status = STATUS.FAILED }) local num_failed_reds = db:count(self.task:get_red_jobs_ns(), { status = STATUS.FAILED }) io.stderr:write(string.format("# Failed maps %d\n", num_failed_maps)) io.stderr:write(string.format("# Failed reduces %d\n", num_failed_reds)) -- self.task:insert{ stats = { map_sum_cpu_time = map_sum_cpu_time, red_sum_cpu_time = red_sum_cpu_time, total_sum_cpu_time = map_sum_cpu_time + red_sum_cpu_time, map_sum_real_time = map_sum_real_time, red_sum_real_time = red_sum_real_time, total_sum_real_time = map_sum_real_time + red_sum_real_time, sum_sys_time = (map_sum_real_time + red_sum_real_time - map_sum_cpu_time - red_sum_cpu_time), map_real_time = map_real_time, red_real_time = red_real_time, total_real_time = map_real_time + red_real_time, iteration_time = total_time, failed_map_jobs = num_failed_maps, failed_red_jobs = num_failed_reds, } } io.stderr:write(string.format("# Server time %f\n", total_time)) -- FINAL EXECUTION io.stderr:write("# \t Final execution\n") collectgarbage("collect") server_final(self) -- until self.finished local storage,path = self.configuration_params.storage:match("([^:]+):(/.*)") if storage == "shared" then utils.remove(path) end end -- SERVER METATABLE local server_metatable = { __index = server_methods } server.new = function(connection_string, dbname, auth_table) local cnn_obj = cnn(connection_string, dbname, auth_table) local obj = { cnn = cnn_obj, task = task(cnn_obj), } setmetatable(obj, server_metatable) return obj end ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- server.utest = function() assert(count_digits(0) == 1) assert(count_digits(1) == 1) assert(count_digits(9) == 1) assert(count_digits(10) == 2) assert(count_digits(99) == 2) assert(count_digits(111) == 3) assert(count_digits(1111) == 4) -- compute real time and compute sum local c = cnn("localhost", "test") local db = c:connect() db:drop_collection("test.times") local min_started,max_written = 10,20 db:insert("test.times", { started_time = min_started, written_time = 16 }) db:insert("test.times", { started_time = 14, written_time = max_written, }) assert(compute_real_time(db, "test.times") == (max_written-min_started)) assert(compute_sum(db, "test.times", "started_time") == min_started + 14) assert(compute_sum(db, "test.times", "written_time") == max_written + 16) end ------------------------------------------------------------------------------ return server lua-mapreduce-0.4.0/mapreduce/task.lua000066400000000000000000000260651263627443400176750ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- Task class sets the task job for the server, which indicates the workers -- which kind of work (MAP,REDUCE,...) must perform. Workers use task objects to -- take a job from Mongo database, and to configure the job parameters. The task -- is related with 'task' collection in MongoDB. This collection is a singleton -- with data necessary to describe the MapReduce task. The following is an -- example of the 'task' collection after the execution of wordcount example: --[[ > db.task.find().pretty() { "_id" : "unique", "init_args" : { "-1" : "lua", "0" : "execute_server.lua" }, "status" : "FINISHED", "storage" : "gridfs:/tmp/lua_sQHr2G", "iteration" : 1, "mapfn" : "examples.WordCount.mapfn", "partitionfn" : "examples.WordCount.partitionfn", "reducefn" : "examples.WordCount.reducefn", "combinerfn" : "examples.WordCount.reducefn", "finished_time" : 1400225799.669539, "started_time" : 1400225797.659308, "stats" : { "map_sum_cpu_time" : 0.039485000000000006, "map_sum_real_time" : 0.07356524467468262, "map_real_time" : 0.08571505546569824, "red_sum_cpu_time" : 0.05029200000000001, "red_sum_real_time" : 0.0801548957824707, "red_real_time" : 0.12635302543640137, "sum_sys_time" : 0.0639431404571533, "total_sum_cpu_time" : 0.08977700000000002, "total_sum_real_time" : 0.15372014045715332 "total_real_time" : 0.2120680809020996, "iteration_time" : 2.0102310180664062, } } ]] local task = { _VERSION = "0.4", _NAME = "task", } local utils = require "mapreduce.utils" local job = require "mapreduce.job" local tuple = require "mapreduce.tuple" local STATUS = utils.STATUS local TASK_STATUS = utils.TASK_STATUS local grp_tmp_dir = utils.GRP_TMP_DIR local get_storage_from = utils.get_storage_from -- PRIVATE FUNCIONS local function tmpname_summary(tmpname) return tmpname:match("([^/]+)$") end local function task_set_task_status(self, status, tbl) self.tbl = tbl or {} self.tbl.status = status if self.tbl.status == TASK_STATUS.MAP then self.current_jobs_ns = self.map_jobs_ns self.current_results_ns = self.map_results_ns self.current_fname = self.tbl.mapfn self.current_args = self.tbl.init_args elseif self.tbl.status == TASK_STATUS.REDUCE then self.current_jobs_ns = self.red_jobs_ns self.current_results_ns = self.red_results_ns self.current_fname = self.tbl.reducefn self.current_args = self.tbl.init_args end end -- PUBLIC METHODS function task:create_collection(task_status, params, iteration) local db = self.cnn:connect() assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = { status = task_status, -- mapfn = params.mapfn, reducefn = params.reducefn, partitionfn = params.partitionfn, combinerfn = params.combinerfn, init_args = params.init_args, -- storage = params.storage, -- iteration = iteration, started_time = 0, finished_time = 0, }, }, true, false) ) self.tbl = params end function task:get_storage() assert(self.tbl) return get_storage_from(self.tbl.storage) end function task:insert_finished_time(t) local db = self.cnn:connect() assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = { finished_time = t, }, }, false, false) ) end function task:insert_started_time(t) local db = self.cnn:connect() assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = { started_time = t, }, }, false, false) ) end function task:insert(t) local db = self.cnn:connect() assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = t }, false, false) ) end function task:update() local db = self.cnn:connect() local tbl = db:find_one(self.ns, { _id = "unique" }) if tbl then task_set_task_status(self, tbl.status, tbl, db) else self.current_results_ns = nil self.current_results_ns = nil self.current_fname = nil self.current_args = nil end self.tbl = tbl end function task:finished() return not self.tbl or self.tbl.status == TASK_STATUS.FINISHED end function task:get_task_status() if self.tbl then return self.tbl.status else return TASK_STATUS.FINISHED end end function task:has_status() return self.tbl ~= nil end function task:get_iteration() return self.tbl.iteration end function task:set_task_status(status, extra) local db = self.cnn:connect() assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = { status = status } }, true, false) ) if extra then assert( db:update(self.ns, { _id = "unique" }, { ["$set"] = extra }, true, false) ) end task_set_task_status(self, status, self.tbl) end function task:get_task_ns() return self.ns end function task:get_map_jobs_ns() return self.map_jobs_ns end function task:get_red_jobs_ns() return self.red_jobs_ns end function task:get_map_results_ns() return self.map_results_ns end function task:get_red_results_ns() return self.red_results_ns end function task:get_jobs_ns() return self.current_jobs_ns end function task:get_results_ns() return self.current_results_ns end function task:get_fname() return self.current_fname end function task:get_args() return self.current_args end function task:get_reduce_fname() return self.tbl.reducefn end function task:get_reduce_args() return self.tbl.init_args end function task:get_partition_fname() return self.tbl.partitionfn end function task:get_partition_args() return self.tbl.init_args end -- TASK INTERFACE local cache_map_ids = {} local cache_inv_map_ids = {} function task.reset_cache() cache_map_ids = {} cache_inv_map_ids = {} end -- workers use this method to load a new job in the caller object local count_idle_iterations = 0 function task:take_next_job(tmpname) local db = self.cnn:connect() local task_status = self:get_task_status() if task_status == TASK_STATUS.WAIT then return TASK_STATUS.WAIT -- the worker needs to wait for jobs being ready elseif task_status == TASK_STATUS.FINISHED then return TASK_STATUS.FINISHED -- all jobs are done end -- take note of the name spaces for jobs and results local jobs_ns = self:get_jobs_ns() local results_ns = self:get_results_ns() -- ask mongo to take a free job by setting its data local t = utils.time() local query = { ["$or"] = { { status = STATUS.WAITING, }, { status = STATUS.BROKEN, }, }, } -- after first iteration, map jobs done previously will be taken if possible, -- reducing the overhead for loading data if self:get_iteration() > 1 and task_status == TASK_STATUS.MAP then query._id = { ["$in"] = cache_map_ids } -- check the count if db:count(jobs_ns, query) == 0 then -- if zero, count one more IDLE iteration, and check the counter count_idle_iterations = count_idle_iterations + 1 query._id = nil -- in case the counter is within the limit, take only a broken job in case -- the counter exceeds its limit, take any job, broken or waiting if count_idle_iterations <= utils.MAX_IDLE_COUNT then query["$or"] = nil query.status = STATUS.BROKEN end end end local set_query = { worker = utils.get_hostname(), tmpname = tmpname_summary(tmpname), started_time = t, status = STATUS.RUNNING, } -- FIXME: check the write concern assert( db:update(jobs_ns, query, { ["$set"] = set_query, }, false, -- no create a new document if not exists false) ) -- only update the first match -- FIXME: be careful, this call could fail if the secondary server don't has -- updated its data local job_tbl = db:find_one(jobs_ns, set_query) if job_tbl then -- reset the IDLE counter count_idle_iterations = 0 if task_status == TASK_STATUS.MAP then local _id = job_tbl._id if not cache_inv_map_ids[_id] then cache_inv_map_ids[_id] = true table.insert(cache_map_ids, _id) end end local storage,path = self:get_storage() return task_status,job(self.cnn, job_tbl, task_status, self:get_fname(), self:get_args(), jobs_ns, results_ns, nil, -- not_executable = false self:get_reduce_fname(), self:get_partition_fname(), storage, path) else -- if self.one_job then ... -- the job (if taken) will be freed, making it available to other worker assert( db:update(jobs_ns, set_query, { ["$set"] = { worker = utils.DEFAULT_HOSTNAME, tmpname = utils.DEFAULT_TMPNAME, status = STATUS.WAITING, }, }, false, -- no create a new document if not exists false) ) -- only update the first match return TASK_STATUS.WAIT -- the worker needs to wait end -- if self.one_job then ... else ... end function task:__call(cnn) local dbname = cnn:get_dbname() local obj = { cnn = cnn, ns = dbname .. ".task", map_jobs_ns = dbname .. ".map_jobs", map_results_ns = "map_results", red_jobs_ns = dbname .. ".red_jobs", red_results_ns = "red_results", } setmetatable(obj, { __index=self }) -- local db = obj.cnn:connect() return obj end setmetatable(task,task) ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- task.utest = function() assert(tmpname_summary("/tmp/blah") == "blah") end ------------------------------------------------------------------------ return task lua-mapreduce-0.4.0/mapreduce/test.lua000066400000000000000000000024521263627443400177040ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] local utils = require "mapreduce.utils" local cnn = require "mapreduce.cnn" local fs = require "mapreduce.fs" local job = require "mapreduce.job" local task = require "mapreduce.task" local server = require "mapreduce.server" local worker = require "mapreduce.worker" local persistent_table = require "mapreduce.persistent_table" local heap = require "mapreduce.heap" local tuple = require "mapreduce.tuple" utils.utest() cnn.utest() fs.utest() job.utest() task.utest() server.utest() worker.utest() persistent_table.utest() heap.utest() tuple.utest() print("Ok") lua-mapreduce-0.4.0/mapreduce/tuple.lua000066400000000000000000000257471263627443400200720ustar00rootroot00000000000000--[[ This file is part of Lua-FCES (https://github.com/pakozm/lua-fces) This file is part of Lua-Tuple (https://github.com/pakozm/lua-tuple) This file is part of Lua-MapReduce (https://github.com/pakozm/lua-mapreduce) Copyright 2014, Francisco Zamora-Martinez Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ]] -- Linear implementation of in-mutable and interned tuples for Lua. It is linear -- because tuples are stored into a linear table. A different approach would be -- store tuples into an inverted prefix tree (trie). Major difference between -- both approaches is that linear implementation needs more memory but has -- better indexing time, while prefix tree implementation needs less memory but -- has worst indexing time. local tuple = { _VERSION = "0.1", _NAME = "tuple", } if _VERSION ~= "Lua 5.3" then -- the following hack is needed to allow unpack over tuples local table = require "table" local function table_unpack(t,i,n) i = i or 1 n = n or #t if i <= n then return t[i], table_unpack(t, i + 1, n) end end table.unpack = table_unpack unpack = table_unpack end -- libraries import local assert = assert local getmetatable = getmetatable local ipairs = ipairs local pairs = pairs local select = select local tostring = tostring local type = type local bit32_band = bit32.band local bit32_lshift = bit32.lshift local bit32_rshift = bit32.rshift local bit32_bxor = bit32.bxor local math_max = math.max local string_byte = string.byte local string_format = string.format local string_sub = string.sub local table_concat = table.concat local table_pack = table.pack -- constants local BYTE_MASK = 0x000000FF local WORD_MASK = 0xFFFFFFFF local MAX_NUMBER = 2^32 local MAX_BUCKET_HOLES_RATIO = 100 local NUM_BUCKETS = 2^18 local WEAK_MT = { __mode="v" } -- the list of tuples is a hash table with a maximum of NUM_BUCKETS local list_of_tuples = {} -- a table with metadata of tuples, indexed by tuples reference local tuples_metadata = setmetatable({}, { __mode="k" }) -- iterate over all chars of a string local function char_iterator(data,j) j=j+1 if j < #data then return j,string.byte(data:sub(j,j)) end end -- converts a number into a binary string, for hash computation purposes local function number_iterator(data,j) if j < 4 then local v = bit32_band(bit32_rshift(data,j*8),BYTE_MASK) return j+1,v end end -- forward declaration local compute_hash -- iterates over all the bytes of a value local function iterate(data) local tt = type(data) if tt == "string" then return char_iterator,data,0 elseif tt == "number" then assert(data < MAX_NUMBER, "Only valid for 32 bit numbers") return number_iterator,data,0 elseif tt == "table" then return iterate(compute_hash(data)) elseif tt == "nil" then return function() end else local str = assert(tostring(data), "Needs an array with numbers, tables or strings") return iterate(str) end end -- computes the hash of a given tuple candidate compute_hash = function(t) local h = 0 for i=1,#t do local v = t[i] -- hash computation for every byte number in iterator over v for j,c in iterate(v) do h = h + c h = h + bit32_lshift(h,10) h = bit32_bxor(h, bit32_rshift(h,6)) -- compute hash modules 2^32 h = bit32_band(h, WORD_MASK) end end h = h + bit32_rshift(h,3) h = bit32_bxor(h, bit32_lshift(h,11)) h = h + bit32_lshift(h,15) -- compute hash modules 2^32 h = bit32_band(h, WORD_MASK) return h end -- tuple instances has this metatable local tuple_instance_mt = { -- disallow to change metatable __metatable = false, -- avoid to insert new elements __newindex = function(self) error("Unable to modify a tuple") end, -- concatenates two tuples or a tuple with a number, string or another table __concat = function(a,b) if type(a) ~= "table" then a,b=b,a end local aux = {} for i=1,#a do aux[#aux+1] = a[i] end if type(b) == "table" then for i=1,#b do aux[#aux+1] = b[i] end else aux[#aux+1] = b end return tuple(aux) end, } -- functions for accessing tuple metadata local unwrap = function(self) return tuples_metadata[self][1] end local len = function(self) return tuples_metadata[self][2] end local hash = function(self) return tuples_metadata[self][3] end -- local proxy_metatable = { __metatable = "is_tuple", __index = function(self,k) return unwrap(self)[k] end, __newindex = function(self) error("Tuples are in-mutable data") end, __len = function(self) return len(self) end, -- convert it to a string like: tuple{ a, b, ... } __tostring = function(self) local t = unwrap(self) local result = {} for i=1,#self do local v = t[i] if type(v) == "string" then v = string_format("%q",v) end result[#result+1] = tostring(v) end return table_concat({"tuple{",table_concat(result, ", "),"}"}, " ") end, __lt = function(self,other) local t = unwrap(self) if type(other) ~= "table" then return false elseif #t < #other then return true elseif #t > #other then return false elseif t == other then return false else for i=1,#t do if t[i] > other[i] then return false end end return true end end, __le = function(self,other) local t = unwrap(self) -- equality is comparing references (tuples are in-mutable and interned) if self == other then return true end return self < other end, __pairs = function(self) return pairs(unwrap(self)) end, __ipairs = function(self) return ipairs(unwrap(self)) end, __concat = function(self,other) return unwrap(self) .. other end, __gc = function(self) local h = hash(self) if h then local p = h % NUM_BUCKETS if list_of_tuples[p] and not next(list_of_tuples[p]) then list_of_tuples[p] = nil end end end, __mode = "v", } -- returns a wrapper table (proxy) which shades the data table, allowing -- in-mutability in Lua, it receives the table data and the number of elements local function proxy(tpl,n) setmetatable(tpl, tuple_instance_mt) local ref = setmetatable({}, proxy_metatable) -- the proxy table has an in-mutable metatable, and stores in tuples_metadata -- the real tuple data and the number of elements tuples_metadata[ref] = { tpl, n } return ref end -- builds a candidate tuple given a table, recursively converting tables in new -- tuples local function tuple_constructor(t) -- take n from the variadic args or from t length local n = t.n or #t local new_tuple = { } for i=1,n do local v = t[i] assert(type(i) == "number" and i>0, "Needs integer keys > 0") if type(v) == "table" then -- recursively converts tables in new tuples new_tuple[i] = tuple(v) else -- copies the value new_tuple[i] = v end end -- returns a proxy to the new_tuple table with #t length return proxy(new_tuple,n) end -- metatable of tuple "class" table local tuple_mt = { -- tuple constructor doesn't allow table loops __call = function(self, ...) local n = select('#', ...) local t = table_pack(...) assert(#t == n) if #t == 1 then t = t[1] end if type(t) ~= "table" then -- non-table elements are unpacked when only one is given return t else -- check if the given table is a tuple, if it is the case, just return it local mt = getmetatable(t) if mt=="is_tuple" then return t end -- create a new tuple candidate local new_tuple = tuple_constructor(t) local h = compute_hash(new_tuple) local p = h % NUM_BUCKETS local bucket = (list_of_tuples[p] or setmetatable({}, WEAK_MT)) list_of_tuples[p] = bucket -- Count the number of elements in the bucket and the maximum non-nil key. -- In case the relation between this two values was greater than -- MAX_BUCKET_HOLES_RATIO, the bucket will be rearranged to remove all nil -- holes. local max,n = 0,0 for i,vi in pairs(bucket) do local equals = true -- check equality by comparing all the elements one-by-one if #vi == #new_tuple then for j=1,#vi do local vj = vi[j] if vj ~= new_tuple[j] then equals=false break end end else equals = false end -- BREAKS the execution flow in case the tuple exists in the bucket if equals == true then return vi end max = math_max(max,i) n = n+1 end -- rearrange the bucket when the ratio achieves the threshold if max/n > MAX_BUCKET_HOLES_RATIO then local new_bucket = {} for i,vi in pairs(bucket) do new_bucket[#new_bucket+1] = vi end list_of_tuples[p], bucket = new_bucket, new_bucket max = #bucket collectgarbage("collect") end bucket[max+1] = new_tuple -- take note of the hash number into __metatable array, position 4 tuples_metadata[new_tuple][3] = h return new_tuple end end, } setmetatable(tuple, tuple_mt) ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- tuple.utest = function() local a = tuple(2,{4,5},"a") local b = tuple(4,5) local c = tuple(2,a[2],"a") assert(a == c) assert(b == a[2]) assert(b == c[2]) a,b,c = nil,nil,nil collectgarbage("collect") -- local aux = {} for i=1,10000 do aux[tuple(i,i)] = i end assert(tuple.stats() == 10000) collectgarbage("collect") assert(tuple.stats() == 10000) aux = nil collectgarbage("collect") assert(tuple.stats() == 0) -- assert(not getmetatable(tuple(1))) end -- returns the number of tuples "alive", the number of used buckets, and the -- loading factor of the hash table tuple.stats = function() local num_buckets = 0 local size = 0 for k1,v1 in pairs(list_of_tuples) do num_buckets = num_buckets + 1 for k2,v2 in pairs(v1) do size=size+1 end end if num_buckets == 0 then num_buckets = 1 end local msz = 0 for _,v in pairs(tuples_metadata) do msz = msz + 1 end return size,num_buckets,size/NUM_BUCKETS,msz end return tuple lua-mapreduce-0.4.0/mapreduce/utils.lua000066400000000000000000000342111263627443400200630ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] local mongo = require "mongo" local heap = require "mapreduce.heap" assert(mongo._VERSION == "0.5" or tonumber(mongo._VERSION) > 0.5) local utils = { _VERSION = "0.3", _NAME = "mapreduce.utils", DEFAULT_RW_TIMEOUT = 300, -- seconds DEFAULT_SLEEP = 1, -- seconds DEFAULT_MICRO_SLEEP = 0.1, -- seconds DEFAULT_HOSTNAME = "", DEFAULT_TMPNAME = "", DEFAULT_DATE = 0, STATUS = { WAITING = 0, -- default job status RUNNING = 1, -- status when a worker is running the job BROKEN = 2, -- a job which is detected as broken FINISHED = 3, -- a finished job WRITTEN = 4, -- a finished job which results has been written FAILED = 5, -- a job which achieves the maximum number of retries }, TASK_STATUS = { WAIT = "WAIT", MAP = "MAP", REDUCE = "REDUCE", FINISHED = "FINISHED", }, MAX_IDLE_COUNT = 5, -- maximum number of iterations without job MAX_WORKER_RETRIES = 3, MAX_JOB_RETRIES = 3, MAX_PENDING_INSERTS = 50000, MAX_IT_WO_CGARBAGE = 5000, MAX_TIME_WO_CGARBAGE = 60, -- 1 minute MAX_MAP_RESULT = 5000, MAX_TASKFN_VALUE_SIZE = 16*1024, -- 16 KB GRP_TMP_DIR = "/tmp/grouped", } local STATUS = utils.STATUS ------------------------------------------------------------------------------- local function connect(cnn_string, auth_table) local db = assert( mongo.Connection.New{ auto_reconnect=true, rw_timeout=utils.DEFAULT_RW_TIMEOUT} ) assert( db:connect(cnn_string) ) if auth_table then assert( db:auth(auth_table), "Authtentication failure") end assert( not db:is_failed(), "Impossible to connect :S" ) return db end local function get_hostname() local p = io.popen("hostname","r") local hostname = p:read("*l") p:close() return hostname end local function sleep(n) mongo.sleep(n) end local function time() return mongo.time() end -- makes a map/reduce job document local function make_job(key, value) assert(key~=nil and value~=nil, "Needs a key and a value") return { _id = tostring(key) or error("Key must be convertible to string"), value = value, worker = utils.DEFAULT_HOSTNAME, tmpname = utils.DEFAULT_TMPNAME, creation_time = time(), status = utils.STATUS.WAITING, repetitions = 0, } end local function escape(str) if type(str) == "number" then return tostring(str) else if type(str) == "string" then str = string.format("%q",str) else str = assert(tostring(str),"Unable to convert to string map value") end -- remove new lines return ( str:gsub("\\\n","\\n") ) end end local function serialize_table_ipairs(t) local result = {} for _,v in ipairs(t) do table.insert(result, escape(v)) end return string.format("{%s}",table.concat(result, ",")) end -- receives a dictionary and returns an array with the keys in order local function keys_sorted(result) local keys = {} for k,v in pairs(result) do table.insert(keys,k) end table.sort(keys) return keys end -- iterates over all the lines of a given gridfs filename, and returns the first -- and last chunks where the line is contained, and the first and last position -- inside the corresponding chunks local function gridfs_lines_iterator(gridfs, filename) local gridfile = assert( gridfs:find_file_by_name(filename) ) local size = #gridfile local current_chunk = 0 local current_pos = 1 local abs_pos = 0 local num_chunks = gridfile:num_chunks() local chunk,data local tbl = {} return function() -- capture the variables to avoid garbage collection, and to improve -- performance local gridfs = gridfs local gridfile = gridfile if current_chunk < num_chunks then chunk = chunk or assert( gridfile:chunk(current_chunk) ) if current_pos <= chunk:len() then local first_chunk = current_chunk local last_chunk = current_chunk local first_chunk_pos = current_pos local last_chunk_pos = current_pos -- faster than tbl = {} for k,v in ipairs(tbl) do tbl[k] = nil end local found_line = false repeat chunk = chunk or assert( gridfile:chunk(current_chunk) ) data = data or chunk:data() local chunk_len = chunk:len() local match = data:match("^([^\n]*)\n", current_pos) if match then tbl[ #tbl+1 ] = match current_pos = #match + current_pos + 1 -- +1 because of the \n abs_pos = #match + abs_pos + 1 found_line = true else -- if match ... then -- inserts the whole chunk substring, no \n match found tbl[ #tbl+1 ] = data:sub(current_pos, chunk_len) current_pos = chunk_len + 1 -- forces to go next chunk abs_pos = abs_pos + #tbl[#tbl] end -- if match ... then else ... last_chunk_pos = current_pos - 1 last_chunk = current_chunk -- go to next chunk if we are at the end if current_pos > chunk_len then current_chunk = current_chunk + 1 current_pos = 1 chunk,data = nil,nil end -- avoids to process empty lines if found_line and first_chunk==last_chunk and last_chunk_pos==first_chunk_pos then -- faster than tbl = {} for k,v in ipairs(tbl) do tbl[k] = v end found_line = false first_chunk = current_chunk last_chunk = current_chunk first_chunk_pos = current_pos last_chunk_pos = current_pos end -- until found_line or current_chunk >= num_chunks return table.concat(tbl),first_chunk,last_chunk,first_chunk_pos,last_chunk_pos,abs_pos,size end -- if current_pos < chunk:len() ... end -- if current_chunk < gridfile:num_chunks() ... assert(abs_pos == size, string.format("Unexpected end-of-file (found %d != expected %d): %s\n", abs_pos, size, filename)) end -- return function() end -- receives a s object, an array of filenames (remote filenames), and a lines -- iterator builder, and performs a merge operation over all its lines; the -- files content is expected to be something like "return k,v" in every line, so -- them could be loaded as Lua strings local function merge_iterator(fs, filenames, make_lines_iterator) -- initializes all the line iterators (one for each file) local line_iterators = {} for _,name in ipairs(filenames) do line_iterators[ #line_iterators+1 ] = make_lines_iterator(name) end local finished = false -- heap for efficient merge operation local cmp = function(a,b) return a[1] < b[1] end local queue = heap(cmp) local data = {} -- take the next data of a given file number local take_next = function(which) if line_iterators[which] then local line = line_iterators[which]() if line then local l,k,v = line,assert(load(line), string.format("Impossible to load line '%s' from '%s'", line, filenames[which]))() queue:push({ k, v, l, which }) else line_iterators[which] = nil end end end -- merge all the data which shares min key local merge_min_keys = function() local top = queue:top() queue:pop() local key = top[1] -- key local result = top[2] -- value take_next(top[4]) while not queue:empty() and key == queue:top()[1] do local aux = queue:top() local v = aux[2] local which = aux[4] queue:pop() take_next(which) for j=1,#v do result[ #result+1 ] = v[j] end end return key,result end -- initialize data with first line over all files for i=1,#filenames do take_next(i) end local counter = 0 -- the following closure is the iterator return function() local fs = fs local MAX_IT_WO_CGARBAGE = utils.MAX_IT_WO_CGARBAGE local assert = assert local data = data local take_next = take_next local queue = queue -- merge all the files until empty queue (finished) if not queue:empty() then counter = counter + 1 -- local key,result = merge_min_keys() -- verbose output if counter % MAX_IT_WO_CGARBAGE == 0 then collectgarbage("collect") end return key,result end -- if not finished end -- return function end local function get_storage_from(str,new) local str = str or "gridfs" local storage,path = str:match("([^:]+):(/.*)") if not storage then assert(new) -- sanity check storage = str:match("([^:]+)") path = os.tmpname() os.remove(path) end assert(storage and path, string.format("Given incorrect storage %s", str)) return storage,path end local function remove(filename) if not os.remove(filename) then return os.execute(string.format("rm -fd %s",filename)) else return true end end local function rename(old,new) if not os.rename(old,new) then return os.execute(string.format("mv -f %s %s",old,new)) else return true end end local function clear_table(t) for k,_ in pairs(t) do t[k] = nil end end local function copy_table_ipairs(dst,src) for i=#src+1,#dst do dst[i] = nil end for i=1,#src do dst[i] = src[i] end end -- checks that value is JSON compatible local function assert_check(value) local tt = type(value) if tt == "table" then local n=0 local not_string_key = false for k,v in pairs(value) do assert_check(k,v) n = n + 1 if type(k) ~= "string" then not_string_key = true end end if n ~= #value and not_string_key then error("Impossible to mix not string keys with string keys") end elseif tt == "function" then error("Impossible to assign a function in a JSON table") elseif tt == "userdata" then error("Impossible to assign a userdata in a JSON table") elseif tt == "thread" then error("Impossible to assign a thread in a JSON table") end end -------------------------------------------------------------------------------- ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- utils.utest = function() local db = connect("localhost") assert( not db:is_failed() ) assert( type(get_hostname()) == "string" ) assert( math.floor(time()) == math.floor(os.time()) ) assert( escape(120) == tostring(120) ) assert( escape("30") == "\"30\"" ) assert( escape("30\n") == "\"30\\n\"" ) assert( serialize_table_ipairs{1,2,3,"hola"} == "{1,2,3,\"hola\"}" ) assert( serialize_table_ipairs(keys_sorted{ c=1, a=2, b=3 }) == "{\"a\",\"b\",\"c\"}" ) -- lines iterator local gridfs = mongo.GridFS.New(db, "test") local lines = { "first line", "second", "third longer line", "a" } gridfs:store_data(table.concat(lines,"\n"), "lines") local i=0 for line in gridfs_lines_iterator(gridfs, "lines") do i=i+1 assert(lines[i] == line) end -- merge iterator local f1_lines = { "return 1,{1,1}", "return 2,{1}", "return 3,{1}", } local f2_lines = { "return 1,{1,1,1,1}", "return 3,{1}", "return 4,{1}", } local result = { {1, {1,1,1,1,1,1}}, {2, {1}}, {3, {1,1}}, {4, {1}} } gridfs:store_data(table.concat(f1_lines,"\n"), "f1") gridfs:store_data(table.concat(f2_lines,"\n"), "f2") local i=0 for key,value in merge_iterator(gridfs, { "f1", "f2" }, function(name) return gridfs_lines_iterator(gridfs, name) end) do i=i+1 assert(result[i][1] == key) assert(serialize_table_ipairs(result[i][2]) == serialize_table_ipairs(value)) end -- for _,storage in ipairs{ "gridfs", "sshfs", "shared" } do for _,path in ipairs{ "", ":/tmp/dir" } do local a,b = get_storage_from(string.format("%s%s", storage, path),true) assert(a == storage) assert(b and (path=="" or (":"..b)==path)) end end -- local tmp1,tmp2 = os.tmpname(),os.tmpname() assert( rename(tmp1,tmp2) ) assert( remove(tmp2) ) -- local t = { 1, 3, a=4, b=5 } clear_table(t) assert(not t[1] and not t[2] and not t.a and not t.b) -- local src = { 1, 2, 3, 4 } local dst1 = { 5, 6, 7} local dst2 = { 5, 6, 7, 8, 9, 10} copy_table_ipairs(dst1,src) copy_table_ipairs(dst2,src) assert(#dst1 == #src) assert(#dst2 == #src) for i=1,#src do assert(dst1[i] == src[i] and dst2[i] == src[i]) end end -------------------------------------------------------------------------------- utils.get_table_fields = get_table_fields utils.get_table_fields_ipairs = get_table_fields_ipairs utils.get_table_fields_recursive = get_table_fields_recursive utils.get_hostname = get_hostname utils.sleep = sleep utils.time = time utils.make_job = make_job utils.connect = connect utils.escape = escape utils.serialize_table_ipairs = serialize_table_ipairs utils.gridfs_lines_iterator = gridfs_lines_iterator utils.keys_sorted = keys_sorted utils.merge_iterator = merge_iterator utils.get_storage_from = get_storage_from utils.rename = rename utils.remove = remove utils.clear_table = clear_table utils.copy_table_ipairs = copy_table_ipairs utils.assert_check = assert_check -- utils.tojson = mongo.tojson return utils lua-mapreduce-0.4.0/mapreduce/worker.lua000066400000000000000000000150251263627443400202360ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] -- The worker is a class which implements the basic procedure executed in a -- client machine. Workers execute map/reduce jobs, until all task jobs are -- being executed. They could be configured to process more than one task. The -- most important methods are "configure" and "execute". Execute method runs the -- job in a protected environment, allowing to capture asserts and Lua errors. -- Broken jobs will be tagged adequately in the Mongo database. Jobs are taken -- from 'map_jobs' and 'red_jobs' collections. Intermediate data is stored in -- gridfs or shared file system or locally and copied using scp. local worker = { _VERSION = "0.2", _NAME = "mapreduce.worker", } local job = require "mapreduce.job" local utils = require "mapreduce.utils" local task = require "mapreduce.task" local cnn = require "mapreduce.cnn" -- PRIVATE FUNCTIONS AND PROPERTIES -- executes the worker main loop; it runs querying the task object for new jobs local worker_execute = function(self) print("# HOSTNAME ", utils.get_hostname()) local task = self.task local iter = 0 local ITER_SLEEP = utils.DEFAULT_SLEEP -- limits parameters, given at worker:configure(...) method local MAX_ITER = self.max_iter local MAX_SLEEP = self.max_sleep local MAX_TASKS = self.max_tasks ----------------------------------------------------------- local ntasks = 0 local job_done while iter < MAX_ITER and ntasks < MAX_TASKS do collectgarbage("collect") local counter = 0 repeat counter = counter + 1 if counter % utils.MAX_IT_WO_CGARBAGE then collectgarbage("collect") end -- synchronize and update task object and MongoDB, and process takes a job -- if possible task:update() local task_status,job = task:take_next_job(self.tmpname) self.current_job = job if job then if not job_done then print("# New TASK ready") end print(string.format("# \t Executing %s job _id: %q", task_status, job:status_string())) local t1 = utils.time() ---------------------------------------------------------------------- ---------------------------------------------------------------------- -- job execution local elapsed_time = job:execute() -- map/reduce ---------------------------------------------------------------------- ---------------------------------------------------------------------- self.current_job = nil print(string.format("# \t\t Finished: %f elapsed user time, %f real time", elapsed_time, utils.time() - t1)) job_done = true else -- if dbname then ... else print("# \t Running, waiting for new jobs...") collectgarbage("collect") self.cnn:flush_pending_inserts(0) utils.sleep(utils.DEFAULT_SLEEP) end -- if dbname then ... else ... end until task:finished() -- repeat self.cnn:flush_pending_inserts() if job_done then print("# TASK done") iter = 0 ITER_SLEEP = utils.DEFAULT_SLEEP ntasks = ntasks + 1 job_done = false job.reset_cache() task.reset_cache() end if ntasks < MAX_TASKS then print(string.format("# WAITING...\tntasks: %d/%d\tit: %d/%d\tsleep: %.1f", ntasks, MAX_TASKS, iter, MAX_ITER, ITER_SLEEP)) utils.sleep(ITER_SLEEP) ITER_SLEEP = math.min(MAX_SLEEP, ITER_SLEEP*1.5) end iter = iter + 1 end end -- WORKER METHODS local worker_methods = {} -- execute wrapper, a public method which runs the jobs in a protected -- environment (xpcall) function worker_methods:execute() local num_failed_jobs = 0 local failed_jobs = {} repeat local ok,msg = xpcall(worker_execute, debug.traceback, self) if not ok then -- in case of error, if a job is available, mark it as broken; additionally, -- insert the error message into 'errors' collection if self.current_job then self.current_job:mark_as_broken() local id = self.current_job:get_id() if not failed_jobs[id] then num_failed_jobs = num_failed_jobs + 1 end failed_jobs[id] = true end -- if self.current_job then self.cnn:flush_pending_inserts(0) self.cnn:insert_error(utils.get_hostname(), msg) io.stderr:write(string.format("Error executing a job: %s\n",msg)) utils.sleep(utils.DEFAULT_SLEEP*4) end -- if not ok then until ok or num_failed_jobs >= utils.MAX_WORKER_RETRIES print(string.format("# Worker retries: %d",num_failed_jobs)) if num_failed_jobs >= utils.MAX_WORKER_RETRIES then error("Maximum number of retries achieved") end end -- configuration of worker, allows to change parameters 'max_iter', 'max_sleep' -- and 'max_tasks' function worker_methods:configure(t) local inv = { max_iter=true, max_sleep=true, max_tasks=true } for k,v in pairs(t) do assert(inv[k], string.format("Unknown parameter: %s\n", k)) self[k] = v end end -- WORKER METATABLE local worker_metatable = { __index = worker_methods, __gc = function(self) os.remove(self.tmpname) end } worker.new = function(connection_string, dbname, auth_table) local cnn_obj = cnn(connection_string, dbname, auth_table) local obj = { cnn = cnn_obj, task = task(cnn_obj), tmpname = os.tmpname(), -- default values max_iter = 20, max_sleep = 20, max_tasks = 1, } setmetatable(obj, worker_metatable) return obj end ---------------------------------------------------------------------------- ------------------------------ UNIT TEST ----------------------------------- ---------------------------------------------------------------------------- worker.utest = function() end ------------------------------------------------------------------------------ return worker lua-mapreduce-0.4.0/misc/000077500000000000000000000000001263627443400152055ustar00rootroot00000000000000lua-mapreduce-0.4.0/misc/digits.png000066400000000000000000000605771263627443400172150ustar00rootroot00000000000000PNG  IHDR@0IaFIDATxڥۖ亪DgUe"f W|N@ vss>ߋ1g@Bm~)ѹ2?':S@&&~|̶OӐR/%ڥOO(KU+`& "r՟zzT_Zo|bjL| X2?ڱZYj_@Up_6CMʇNKwH'HL#$B([6ڈ|so|q[%q;L4{fҊojg;Wri^Y1k0?ro}@`= dV@/ dM&|ˇ{_qStuG[9p˾v[tw6rRv sZ`rxBS ։Q+2nUzQ oH=Od_pXs"R502|=!Ny_| "7 /5Rݲ { x\gטo*;esΩ3%O.8a7-yf,1jan[k~0B,S,*!7@9 iia@>*Jxs] I $N7@QFm9 Ͷ-$xߎQQ4yPVs<^.jP  6s ~B%Eչ P0? )mDp|+Wpi$vY~ 艕[UN@逞,9ic ]xEQQ/٣i?hG))(/Ebd,#Ahmp. Ȧ#K_6X/| f"{"wa>hݠ.H#7p, `c}Eʭ _-!6UZN> \|ͺKСAY U^6BE3:>5c}6V/w& ,<]x!HHr s"ȳ]E>YR 䌢PU ֯gU 8V .A>HaΎPmKt* ) P*sbԟG!1;faomSKtTsEHVx7?~m ׄ'@lSڽ&Ƅq@_>UHaN>,\/,, ~=> ߥDAh4w={z +EJ5-J0w6)3':~ND~0 fwYWbb "TH1 q3ҥ᛾ dXSCJ'䲂\[;I¢nQ!; F&A )RSWt!Kpn'R6yrņn>(bB i($q} ӭ 6&š< :!ۻJ%@JWb:0[Eg Q_e iaX)!'y&c6 5!5~şRfs⟋ O$LK`>o?Ȟ_d?W>K+C|AR 0@P-?DÚo`)\Qf# ~y2J _H0% g9$< 5wz6y< Ookv g 3l8SQH XdmjE uuT,F(0[RyZ2`3KD4CU?g m*L%WO,B^/x-Mۃu4;zG TnX:3:]kkB?4Ds.Ҭ Dr맽Qᰉcg_8wHw@яaE0][\㗀kYr6rCx7-}XBz%DT߳zbRb:#>f0%qN=֕#`ј9I "' U%$R[OHs a)/ډ}F%%oۚjZ}=_LAb TD H_ml /۝b{K|W$6l2< J[)ԴL@k:‰q/PQ!w@JfQDbSH;y(ʀj(ǂ8p4 ?6e oa=^֜z `Od bɤ~!:=W 3\ 9B'U~_}$yQɉ.߯ Q]!1ν cFusI"SײO1G)Kgˀԁ +rwȢ&TniC'~p+ J5Jj=1mi,gp?1MwF~e#_x}4xFM(go9VMo,K0Y@|t E=\cCl7|Cu_2|.Dg@.[h|L]@;pXc)ĔI%§ vR_WFbA|?z7+w@[vq6m%OR {@~85(g Kڡ2T[xY_TDAX^~smX '3;i]%h/mLXhxe]4JLʐ/`Z.XpW8'eYpVHzR3 XJɀė,9ꌏ0"1 ?Ch_є!D5x8w |TcC 8 ,)26S,o;r}lR+lm-eTdugTW'(p#4iZ*=!f $:M w|.'\-n1`n5}C]MTN@ D@c2 ȗ_GbaѵC郝yrcP_X7S<9V:(K #,m0nn<?O<sF~O0S*O:~KdL0yzGp^NT"A|c.I*VM-t|U |i?(+iXE?_ /KިsKÄ,*f.g^E޾猴'7zuXcw㷖Rh1>LOj%ʇN 0r.` ZV>8MZEO|!u `eW!-OfNy/}ttw\V'>>}TL` CmŒE<HtI&(P1 pAi4lk d\H*pQR'@hhQfaDW*8 Btz\6,A,I?^-|jЮs$_W&eF/`f `!uv^[#MŤ},J.p(dR";[.u'5r7 UeȀ$8-}_Sb)V} )#? 2ޑbN$}̋U#=E`.DnA AfnI_KFaܰ.cn$Z8=K3ZMx̐|)31Jo32<|5WAIB;$(Uyox RV2pȼWyIr~%hy "1uC?ƀ]bEG80&sK;4LŊvkǤaXoL#`/$9 /?sTN2̈MiV1ZI)'.U)aQOXIIC| H.[{\&*$ _ ^~zc[̉$B7(~5 Ѓu –`+yG6lt҃!Tu?L1Z˔5|aW|E^ `;ox&/5Q4yl58=8jGP?z4 dpt88=  HB\VhҶ}dgy81Xd# (o9W^ysRᯉ X 84a7>]lxSr@xi2-zV'b&gZ"OZg 7R@78NoPˈ /C&(qVB| ?;y=4ߧL: Z6]\D"GI BiͰ^|sSu5v@NR1P`OPpn#nlKPY$vLb+#Q0Hй{,Gt&˯B/|_AgVZxܚPj r{l՚<)\Tt]P0A*aA *77Y"4OH|U?Usps,B SƟ$=;žز=fŵf5vmn6)@Aj2 >R!vX+m@KwƵP- ^%8n 7 M5H"Z `e@Cs6.*㡁 XVI[I$*Ǩr_,F%@ x;TTf3T/9YATl `IFLwMp /D'$" ]{R'Q&,KohSH#3 ’׀5?6-Hd.y3DZti>@HSH T7t@N ?hA$_001Q8| .rlk,M^[#P6j?x6\8>p(fo|ijrb| yK< `mzKk m {k[~QTќ\haڲ>{Wr0dwIDpj?(R/\T# S +EkЬKVzz|(,^x%@9<ULx Ŋ)pAj 5(1F.0a S!$!{%b|=KC0OqWzF+!-^"֞wDi5%3(cm1ӗe<bA<`h/,Bvs PH|X8޽r1 J2^ QB_S"%c>E@#Ňq<@PA1ky@t@fȯ% :g^/,ۼy?nX+!sBSTJY_BFn(3R>X#Xz,lgNkk}Kt򩈷"Ѕ=X~evs|v4颧F\؝+V }e@;V=&){kf2=4jۋZOViOxBo=iȋksF nb#{,;1"3 Rý>#)yGkxlS< *g@ވ;YT32<IlQFI `57R!б t!!Ҟv0q ؀@l{_$|>e%@x+P'ԏ%G|Y4W[qz %nQAMW F1 hy Ahj/kcovہR66Ĕ%#5@AZ;̒)>"%9d^q;c@UW =G$}4U!d>t3}GTV`4_? q@`ӿ!®) c 0eIR!V[μdhZC0#ҶS4N?@fQ{wcpa'/TUт7Q$$@3 25`!quNLʊ%>|,hfkfp\zLr M@3NA9f٬r+%9 xvjB@݈;r>j)؍ _4QSK+6{Qg*29y~=byxgg-G\{kuR9Ӹט7dERPc F V;[ hsxLs.3.tz3$X<}̯%MPZؖ`go^5.6ԐD@~ًf-Y#Ba粚Jްk} ~x;9>˓׳ `Y;hoAd@w"< I$:ڽ}SV@).*ӣ&#!"Wi?fk\oUV@5nPj4[Ūpp^$xYxFTC=@u,~d@6"ER~Zx_QRVlz;a5&b2 rsEw\$'ۅK%P^&Q1^pF@c  *Rq}׳ K'2h>bj2ßyX1\ZJ0+%a8%ܼL{Lbʪ|sߣ+۶iǁAj;M$Ԉ>0FK!eʃV644aq'k|l<(& L@ :nܔuL.GVKp$Ue;Xg O|V @"j+7 .g@e:`@pzwx:a^d eֵI^s>O"P 9uPzK&׉m+pO-{؝_R=!x;YgזՁIvQS a'S``}" /2[#_JCUV c5UM+]ۭ($LuYÌyPMzP dfc4zJB ߵeʍ9)~r}k#BZ ! y[5kdW}2a[̈́~5OZ7ٺKo0嘽e?"ab]Z]GA,Xh:,.=m ]zN\! hls0`$;)#/`1:VV9-:'b>{*Q09,!u1TffF1T^Y??UlZp&aadpP=XvU_5Ett#߶jTX [Ѯ`ʉq_ MK8ſK.'gP hn.oLGPP2@]sQ73do:"]`yrTGYE2kt%|z! BLyPA `cm|*z`~2| CSI@p_pj`] (e\F8$ \c LTDbJYM3*Za}Ӵy-^<(>/*[%Hks9v2(pLA- NiiOXd vg(@!, fuallr ^ is*/`TKE1JTF{ڼ1Wr's;%B]=w-NΟK4D!Bjp_7mtQڹy(3@VG#m̧6& }|goT|'Ư-4u}0T'AswO`E@jNx&cB8AlK[Ki|4΅X= aYZD#kQټ3'F'!(rX? fWJIBp\1_sRèzJ3V̫\23^P ~/3j>: U>P f.2ԔOu pSBZ|;B8m \&F>xw_MT{ I=kAnFF/|I-@FDc !^ mc۰FohIEr_NqWB; P'fsj*EKz 8L st@wYe8;!bWکP5,}>X4y7>885 o4@+*e'EXAI+oTt{c  (0nMAo'}3H6p$hmD߬YD;\لΙv?X2t IP6A|-Ǜd zjxc>N&;M"hH-,>v4@XoKL@0;@Ypp~7o#x'> !n;94@(LYc`HUA8CN/sű^t$ -5JcZs&<>E|"OZlr$@= pI:N~x7s'wҴ(nMN Ӹs|ÏtV2[~p&NqBۍ>AQr7'xn3QdL[(q H6nLOY,JSVۭz2S߷7@_nj`.z$hwx+`Ⳍ&kg!XX3q4A]- HaZ"``OL$3 G+Xጙ*!GFDJ>_1_zwbL(q^ۙKp.ozgTqޙKYn.-׷\Jp븧'|"RvBw>^c|ְڤ\ab%ʀ/E47 4xj3|Ow8y&E[/jP@zO\sO@8/A<`~q=m&8@o~zW+>2l8}m棣eD@c9l0 V}d {TX>sAmʀsc(|,Kf8~٪DA,,r?T8C7!m䃻OTTRi$i+/!i%XdEL'@Ʒy?> |@~\zXׯ D|Ů@ /8gK M|7rΕ00"t]0#tȡ~}6bGu!i2n \Q* >LЦ(3 wAsm=O G51Mϊg+p: O0Ь/hAV7axj 6>(_zHKH8go??i`#` _j_ o*@F|zxP3R~q!d!i_dЁE0Z ݑ1D.v22gN/CiIm|m" 8, UQ(8=|q@X2z7lh|i`7H+ /Ou'PF5Ի/@<-!5,}ha'@gӶ<7 ȊY .OÀnWa@eYt_l~FyoK|9AvЈPq`~#B4)`.Of|x {!+ 5~FdN3M%OI y>4E@9nk?E/!zfD?_ \W2}KѤEXRu~i=\O {Kz^WƮ |oMMIo\5L~7_:Ti6;de_) MPUB0YUm d%V6K;I&6@MHZ* -|je@u*D `l>F(;'o `ns/LoN%J-`Ǿ< HH-y?aMcPڂ&UX%?_~-df^xE(6L 82>ǘ h"l"Eע9ֿ5Ao|K?xmm>ݠCl^khbl&,6 ]0M_J0B׍@R+EЯΧ _ik(JxOC4&Ufgj&$)|~FG A;5} iY\I x0`_mNW<*qks%끟l/Yg"-@o&J &@1w9]#G`W:% 65ɭ٪ߞW\a[4"g-ވ 'xGF<K{cl >w𳭅Y6=սEdϺ?"K}p^>kn; ?q'pl9Im3mLĜ}^%YG0`K Xb]EIF=Cl‹5r)o rcj*v~ݜ-Nj} >ey5ǼCyֳ ƜI] PԈ> ax/W'og &$-θ% 4kB^'> +/6A!4'm@- +eO_L|wXŸ/SF ZJmn{jpR$@ ݙK+" pHv;xR͑|W@.^1?|b))1bG?]%_LH x^R2 8-'Jޏ֪8|Rwawiˍpd02gI`$5iϏ1hBDl+9Ԣ=!M~09(YZQ_dI-,cP[p0اPH9" !Ks9—^b6> U?/W; \D>~W#c7@ZLb&#MD&w'qc.5s[#}1(sO0zϸ񵽳ɜֲ nV p ! PagHn4Br ##5^W(e`[ xT0l%,= ش` `2I/@n @;#^ P<%4 1w&8/ggCиʦ'mfՅ Fإ fS`E!wb9EtXwk7+xg\v⥽p|!Z=iHS.JKwKuJYEOE@QHnƃ1@ TС,+&|LhFqTh`юD A k!>}[ k0&~~V@0𥪚No$Gc5.# g)b~XL5=_%g B<::]-o8<8U 77Q֣Psn1YtiώKrl/BS|Фw^3 ئpn;zG0.LwуBIU~x* g[A4KHP2a1;P @fqR4 rːT@.IJP2[ ۚ |dO7!aP# |"@$Q*ǀ|, 1ih&_"4LHo~;8YZ -H%%ʒlASīp OB$]}ѓTPzX A 0|d?7f97mE"y^G= Px"Դ1 mK>"}pGe _/b4bSIw|!o|ExO907}1o x J ,990 c>1a,%٩"9ˇ*Sl:u ! .+L^֪ez/'yȨB^sx߾G L M| ߒSj;=7¥bͼd 2o<ZO3+&@|?+i47 |D@>:0fz[MSJ|AL);\[FG5͠;!s1XqYPn[׮ՓQ.Wy9DLK`RRQLƚzfp>=H- 7c?ۿBի-1{kk~/۪̇G.#C Co(F'. 1; y_fX8N˓'.  1JU+mZ@RT7|k |dw@>?8)7vI<#ﲣ0r Yl?7SEcz`ǯP;M(-|LcȪqzQhtL)& w;`PzFѨvfѴ} Z|yG?8-ן"8#FgT+‡oQt]0KHz) Э6;P NBE; 6E4ߠL?M6VW< DR(_nʽys 5s X$i,iܲ+Zz8vpc~:CAkNOf@WY ]}! je9GzsY@ffo="ksP'`O;e)%*9pX9R$Bzm^2R@sxn%"l @.QKx 92>O'd|C0fV|3Os}n(>&2&QZmHE5AT:CKC  CyA'~1T!}D~gӇ+ |K|Խ|c4˖b27^0JPFc=6wRJ ~>J^ 1D]%)`ֆUgFK@p!zM |^PG{aߖrP5#i77bڽ9i>,ѕ$:bLzeY2/_ft,V>>71koc{G>1-m+D1M#62ǤM0TL1Oņ;i <^kCB|ohԦMOuM֗Ȭҍo|ϧI@R7y\Yg2.] bH_(xpnE -Sy[E8a[2c"2i>n.mTXbG|rckO:Nk7=Bt~#pH :Y|q$?<= LF࣒LB͟KlC][ٞw?0{+/F.9/S< u,"Ƹ6 N PF;{ ]Q c .7E}>3\g@pBMm|䚹9 HY N PA$|_ RW@`֏5RQ}#$)а&ڽWr ҵ4PhqdF7|~/m2/IyEBx@2p^C*z"Y"OLdM|kK!=`@?J@{otTÜ`H2`)v1\T@kϣ A ŷŋtu"H6{$zd]=338.;dovKEF1*ʷv[݂4!n=9"HB_ @ X*3vi%m[~%5ԏ(<6E.^2Pn/lKu~a*8{5$m 6։E?DBϧQz aܤnVv!}Q)#`U,K.f b:bň pCx9 )Il^B/:3bq9XxV ŀfls_tg$u#8# 8` -np$>| zmczN .V(*vg<(=yT1ah-Apv`]K?a /r)b̙<*#bڲ uУf2wzπ?N ̈́f gu[=!0]4ǀkV-lJ+@R|:`0TDV Nm,uO}[MH"@7q_%*twWC2'KU, Ɋ O*Ɍeե$C `"H39ep.@qw$*L uq|FT Tw(Ž!"}h-lXN^.a"{G kGGy?@:`;H1QBrvY\o / `J |{(^ݟ O7:WF}|H KbVZ\4O)Mp؃e4CPoѴnFU.2[_t%ls3RY7}| `v=S])I3 1IX xmBq0BPDl.p\7x ޑpa@0U8 v/R\hP.AG Ao'OV  ˌH<|$@43w|,05 n@T_&@I17ϡb"##*0D=Cٌ R<.V.7ʋH[k)p@:ti EA"c[hX2.=_yƍq 5[bNgO.i;I8껝` I O H)@{" _o#/H^͛ˈ,5`/'GZ &5ְ60\q*:$|;-H 0; "y3tAWFH"I-VUy`&>?a+,n%%\z[^A@v 7OBb!?!tYy~Ld3?#H1l>jju#_&??(N bK`OX8F*USGNE*[[ĶPZ6T@ F)&w9<'Y "1~%kBՊF꛽XBcPoo붘O:y\M񅈞s6_͉{ӵ-OfD7倸%Ѐx<UgW <] 3˾i,9ΧM`PynD3g2)038iܗ]:C\4 8X xhZ?=}@'puBx<]p5}y!5-P, HVMѯFySLv_#`z#t|ϤO*ɩ| oL;Xp~g}#ȢA3qs$uPU_>)i~ÀR/Y.)En;(+ht@Cb r,ѭf^4X}4Ep!AɧQEXpJ&#AEAppǙ<[Ģ~Vh Ss@kp'IWhH-. ț"qL!m0ld ^c4m.hlj˯D{na:D官/bzIXݍ&)9&al cO9TK)Gʨ1hpD`>I\NN_)5>z_D &h[͐7ɯ/&11+ K X:34w33eI"O|kKM|WQx`{-&O)9>Jq>;8PR^ua\"Xc`_BrXūw`F^lsmZ*yOA'vR' =gx*󍣚[$8q9w?kȗMn Q~;QX!j[tx *2gp]զer;U(Rl3w1@XPETWhQ@I;Vnx,Eߪr rf /}|Ёvku?+</- v@K10h޵[ g]zL݌Cp{OK qӏ 0spxL | ~ڋ5B&g\`)G@=Y *9π)c&ėI~AL|9XKS|ro&qM'_ԢXM韼%E#!$գ0YwAYV?Zf!U@~@r/l=Ʒy )аA< I ȅ[ 5CG;˳:(Az\fL3PPc_4e{Hd'h)75yބv"wDN xvR?ΙRh"s>ްd/i^w9R-nn58/=džjI_HE͟_zÊ߭xodb{n?d|IENDB`lua-mapreduce-0.4.0/misc/make_sharded.lua000066400000000000000000000044621263627443400203250ustar00rootroot00000000000000--[[ This file is part of Lua-MapReduce Copyright 2014, Francisco Zamora-Martinez The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ]] local connection_string = arg[1] local dbname = arg[2] local auth_user = arg[3] local auth_dbname = arg[4] or "admin" assert(connection_string and dbname, "At least connection string and dbname are needed as arguments") -- function read_password(msg) io.write(msg) -- enter raw mode os.execute("stty -echo raw") local pass = {} repeat ch = io.read(1) if not ch then break elseif ch == '\r' or ch == '\n' then io.write('\r\n') elseif ch:byte(1) == 127 then io.write('\r') io.write(msg) io.write((' '):rep(#pass)) table.remove(pass,#pass) io.write('\r') io.write(msg) io.write(('*'):rep(#pass)) else table.insert(pass,ch) io.write('*') end io.flush() until ch == '\n' or ch == '\r' or not ch -- activate echo and exit raw mode os.execute("stty echo cooked") -- return table.concat(pass) end -- local mongo = require "mongo" local db = assert( mongo.Connection.New() ) assert( db:connect(connection_string) ) if auth_user then local auth_password = read_password("password: ") assert( db:auth{ username=auth_user, password=auth_password, dbname=auth_dbname }, "Impossible to authenticate with the given credentials" ) auth_password = nil end collectgarbage("collect") local gridfs = assert( mongo.GridFS.New(db, dbname) ) assert( db:run_command("admin", { cmd="enableSharding", enableSharding = dbname }) ) assert( db:run_command("admin", { cmd="shardCollection", shardCollection = dbname .. ".fs.chunks", key = { files_id = 1 } }) ) lua-mapreduce-0.4.0/misc/naive.lua000066400000000000000000000002461263627443400170140ustar00rootroot00000000000000local vocab = {} for line in io.lines() do for w in line:gmatch("[^%s]+") do vocab[w] = (vocab[w] or 0) + 1 end end for w,v in pairs(vocab) do print(v,w) end lua-mapreduce-0.4.0/remove_results.sh000077500000000000000000000002111263627443400176610ustar00rootroot00000000000000#!/bin/bash ns=$1 if [ ! -z $ns ]; then echo "use $ns db.dropDatabase() " | mongo else echo "Needs a namespace as argument!!" fi lua-mapreduce-0.4.0/test.sh000077500000000000000000000052201263627443400155670ustar00rootroot00000000000000#!/bin/bash lua mapreduce/test.lua if [[ $? -ne 0 ]]; then echo "ERROR" exit 1 fi for storage in gridfs shared sshfs; do echo COMBINER + ASSOCIATIVE COMMUTATIVE IDEMPOTENT REDUCER screen -d -m ./execute_example_worker.sh diff <(./execute_example_server.sh $storage | awk '{ print $1,$2 }' | sort) \ <(cat mapreduce/server.lua \ mapreduce/worker.lua \ mapreduce/test.lua \ mapreduce/utils.lua | lua misc/naive.lua | awk '{ print $1,$2 }' | sort) > /dev/null if [[ $? -ne 0 ]]; then echo "ERROR" exit 1 fi echo NO COMBINER + ASSOCIATIVE COMMUTATIVE IDEMPOTENT REDUCER screen -d -m ./execute_example_worker.sh diff <(lua execute_server.lua localhost wordcount \ mapreduce.examples.WordCount.taskfn \ mapreduce.examples.WordCount.mapfn \ mapreduce.examples.WordCount.partitionfn \ mapreduce.examples.WordCount.reducefn \ mapreduce.examples.WordCount.finalfn \ nil $storage | awk '{ print $1,$2 }' | sort) \ <(cat mapreduce/server.lua \ mapreduce/worker.lua \ mapreduce/test.lua \ mapreduce/utils.lua | lua misc/naive.lua | awk '{ print $1,$2 }' | sort) > /dev/null if [[ $? -ne 0 ]]; then echo "ERROR" exit 1 fi echo NO COMBINER + GENERAL REDUCER screen -d -m ./execute_example_worker.sh diff <(lua execute_server.lua localhost wordcount \ mapreduce.examples.WordCount.taskfn \ mapreduce.examples.WordCount.mapfn \ mapreduce.examples.WordCount.partitionfn \ mapreduce.examples.WordCount.reducefn2 \ mapreduce.examples.WordCount.finalfn \ nil $storage | awk '{ print $1,$2 }' | sort) \ <(cat mapreduce/server.lua \ mapreduce/worker.lua \ mapreduce/test.lua \ mapreduce/utils.lua | lua misc/naive.lua | awk '{ print $1,$2 }' | sort) > /dev/null if [[ $? -ne 0 ]]; then echo "ERROR" exit 1 fi echo INIT SCRIPT screen -d -m ./execute_example_worker.sh diff <(lua execute_server.lua localhost wordcount \ mapreduce.examples.WordCount \ mapreduce.examples.WordCount \ mapreduce.examples.WordCount \ mapreduce.examples.WordCount \ mapreduce.examples.WordCount \ mapreduce.examples.WordCount \ $storage | awk '{ print $1,$2 }' | sort) \ <(cat mapreduce/server.lua \ mapreduce/worker.lua \ mapreduce/test.lua \ mapreduce/utils.lua | lua misc/naive.lua | awk '{ print $1,$2 }' | sort) > /dev/null if [[ $? -ne 0 ]]; then echo "ERROR" exit 1 fi echo "Ok with storage $storage" done