Disco Documentation¶

Batteries included¶

The Disco core is written in Erlang, a functional language that is designed for building robust fault-tolerant distributed applications. Users of Disco typically write jobs in Python, which makes it possible to express even complex algorithms with very little code.

For instance, the following fully working example computes word frequencies in a large text:

from disco.core import Job, result_iterator

def map(line, params):
    for word in line.split():
        yield word, 1

def reduce(iter, params):
    from disco.util import kvgroup
    for word, counts in kvgroup(sorted(iter)):
        yield word, sum(counts)

if __name__ == '__main__':
    job = Job().run(input=["http://discoproject.org/media/text/chekhov.txt"],
                    map=map,
                    reduce=reduce)
    for word, count in result_iterator(job.wait(show=True)):
        print(word, count)

Disco is designed to integrate easily in larger applications, such as Web services, so that computationally demanding tasks can be delegated to a cluster independently from the core application. Disco provides an extremely compact Python API – typically only two functions are needed – as well as a REST-style Web API for job control and a easy-to-use Web interface for status monitoring.

Disco also exposes a simple worker protocol, allowing jobs to be written in any language that implements the protocol.

Distributed computing made easy¶

Disco is a good match for a cluster of commodity Linux servers. New nodes can be added to the system on the fly, by a single click on the Web interface. If a server crashes, active jobs are automatically re-routed to other servers without any interruptions. Together with an automatic provisioning mechanism, such as Fully Automatic Installation, even a large cluster can be maintained with only a minimal amount of manual work. As a proof of concept, Nokia Research Center in Palo Alto maintains an 800-core cluster running Disco using this setup.

Main features¶

• Proven to scale to hundreds of CPUs and tens of thousands of simultaneous tasks.
• Used to process datasets in the scale of tens of terabytes.
• Extremely simple to use: A typical tasks consists of two functions written in Python and two calls to the Disco API.
• Tasks can be specified in any other language as well, by implementing the Disco worker protocol.
• Input data can be in any format, even binary data such as images. The data can be located on any source that is accesible by HTTP or it can distributed to local disks.
• Fault-tolerant: Server crashes don’t interrupt jobs. New servers can be added to the system on the fly.
• Flexible: In addition to the core map and reduce functions, a combiner function, a partition function and an input reader can be provided by the user.
• Easy to integrate to larger applications using the standard Disco module and the Web APIs.
• Comes with a built-in distributed storage system (Disco Distributed Filesystem).

I tried to install Disco but it doesn’t work. Why?¶

See Troubleshooting Disco installation. If the problem persists, contact Disco developers on IRC or the mailing list.

How come ssh localhost erl doesn’t use my normal $PATH?¶

ssh localhost erl

is different from:

ssh localhost
erl

In general, interactive shells behave differently than non-interactive ones. For example, see the Bash Reference Manual.

How do I profile programs in Disco?¶

Disco can use the Profile module to profile map and reduce tasks written in Python. Enable profiling by setting profile = True in your disco.job.Job.

Here’s a simple example:

"""
Try running this from the ``examples/faq/`` directory using:

disco run profile.ProfileJob http://example.com/data | xargs disco wait && xargs disco pstats -k cumulative
"""
from disco.job import Job

class ProfileJob(Job):
    profile = True

    @staticmethod
    def map(entry, params):
        yield entry.strip(), 1

How do I debug programs in Disco?¶

Set up a single node Disco cluster locally on your laptop or desktop. It makes debugging a Disco job almost as easy as debugging any Python script.

Do I always have to provide a function for map and reduce?¶

No, you may specify either map or reduce or both. Many simple tasks can be solved with a single map function, without reduce.

It is somewhat less typical to specify only the reduce function. This case arises when you want to merge results from independent map jobs, or you want to join several input files without going through the map phase.

See also: Data Flow in MapReduce Disco Jobs

How many maps can I have? Does a higher number of maps lead to better performance?¶

In theory there is no restriction. In practice, the number is of course limited by the available disk space (for input files) and the amount of RAM that is required by the Disco master. Disco includes a test case, in tests/test_50k.py that starts 50,000 map tasks in parallel. You should be able to add a few zeroes there without any trouble. If you perform any stress tests of your own, let us know about your findings!

Each map and reduce instance is allocated exclusive access to a CPU. This means that the number of parallel processes is limited by the number of available CPUs. If you have 50,000 map instances but only 50 CPUs, only 50 maps are run in parallel while 49,550 instances are either waiting in the job queue or marked as ready — assuming that no other jobs are running in the system at the same time and your input is split to at least 50,000 separate files.

The number of maps can never exceed the number of input files as Disco can’t order many maps to process a single input file. In other words, to run K maps in parallel you need at least K input files. See Pushing Chunked Data to DDFS for more on splitting data stored in Disco Distributed Filesystem.

In general, the question about the expected speedup when increasing parallelism is a rather complicated one and it depends heavily on the task at hand. See Amdahl’s Law for more information about the subject. However, unless your tasks are so light that the execution time is dominated by the overhead caused by Disco, you can expect to gain some speedup by adding more maps until the number of maps equals to the number of available CPUs.

I have one big data file, how do I run maps on it in parallel?¶

See Pushing Chunked Data to DDFS.

How do I pass the output of one map-reduce phase to another?¶

Many algorithms can be implemented cleanly as a sequence of mapreduce jobs. Chaining jobs together is also efficient, as the job’s results are readily distributed and stored in Disco’s internal format.

Here’s an example that runs ten jobs in a sequence, using outputs from the previous job as the input for the next one. The code can also be found in examples/faq/chain.py. The job increments each value in the input by one:

from disco.job import Job
from disco.worker.task_io import chain_reader

class FirstJob(Job):
    input = ['raw://0', 'raw://0']

    @staticmethod
    def map(line, params):
        yield int(line) + 1, ""

class ChainJob(Job):
    map_reader = staticmethod(chain_reader)

    @staticmethod
    def map(key_value, params):
        yield int(key_value[0]) + 1, key_value[1]

if __name__ == "__main__":
    # Jobs cannot belong to __main__ modules.  So, import this very
    # file to access the above classes.
    import chain
    last = chain.FirstJob().run()
    for i in range(9):
        last = chain.ChainJob().run(input=last.wait())
    print(last.name)

Assuming that the input files consists of zeroes, this example will produce a sequence of tens as the result.

How do I print messages to the Web interface from Python?¶

Use a normal Python print statement.

Internally, Disco wraps everything written to sys.stdout with appropriate markup for the Erlang worker process, which it communicates with via sys.stderr. See also The Disco Worker Protocol.

My input files are stored in CSV / XML / XYZ format. What is the easiest to use them in Disco?¶

See disco.worker.task_io.input_stream().

For CSV files you can also have a look at the csv module shipped in the Python standard library.

Why not Hadoop?¶

We see that platforms for distributed computing will be of such high importance in the future that it is crucial to have a wide variety of different approaches which produces healthy competition and co-evolution between the projects. In this respect, Hadoop and Disco can be seen as complementary projects, similar to Apache, Lighttpd and Nginx.

It is a matter of taste whether Erlang and Python are more suitable for the task than Java. We feel much more productive with Python than with Java. We also feel that Erlang is a perfect match for the Disco core that needs to handle tens of thousands of tasks in parallel.

Thanks to Erlang, the Disco core is remarkably compact. It is relatively easy to understand how the core works, and start experimenting with it or adapt it to new environments. Thanks to Python, it is easy to add new features around the core which ensures that Disco can respond quickly to real-world needs.

How do I use Disco on Amazon EC2?¶

In general, you can use the EC2 cluster as any other Disco cluster. However, if you want to access result files from your local machine, you need to set the DISCO_PROXY setting. This configures the master node as a proxy, since the computation nodes on EC2 are not directly accessible.

ssh MASTER -L 8989:localhost:8989

Disco 0.5.3 (TBD)¶

Disco 0.5.2 (June 24, 2014)¶

Disco 0.5.1 (April 16, 2014)¶

Disco 0.5 (February 14, 2014)¶

This release is dedicated to the memory of Priya Hattiangdi, Prashanth Mundkur’s wife, who has passed away a few days ago. May she rest in peace.

Disco 0.4.5 (Mar 28, 2013)¶

Disco 0.4.4 (Dec 5, 2012)¶

Disco 0.4.3 (Aug 22, 2012)¶

Disco 0.4.2 (Apr 26, 2012)¶

Disco 0.4.1 (Sep 23rd 2011)¶

The official Disco repository is now at http://github.com/discoproject/disco

Disco 0.4 (May 4th 2011)¶

Disco 0.3.2 (Dec 6th 2010)¶

Disco 0.3.1 (Sep 1st 2010)¶

Disco 0.3 (May 26th 2010)¶

Disco 0.2.4 (February 8th 2010)¶

Disco 0.2.3 (September 9th 2009)¶

Disco 0.2.2 (July 26th 2009)¶

Disco 0.2.1 (May 26th 2009)¶

Disco 0.2 (April 7th 2009)¶

Background¶

You should have a quick look at Technical Overview before setting up the system, to get an idea what should go where and why. To make a long story short, Disco works as follows:

• Disco users start Disco jobs in Python scripts.
• Jobs requests are sent over HTTP to the master.
• Master is an Erlang process that receives requests over HTTP.
• Master launches slaves on each node over SSH.
• Slaves run Disco tasks in worker processes.

Prerequisites¶

You need at least one Linux/Unix server. Any distribution should work (including Mac OS X).

On each server the following are required:

• SSH daemon and client
• Erlang/OTP R14A or newer
• Python 2.6.6 or newer, or Python 3.2 or newer

The same version of Erlang and Python should be installed on all servers. The default version of Python on the clients from which Disco jobs are submitted should also match that on the servers.

Optionally, DISCO_PROXY needs one of

• Lighttpd 1.4.17 or newer
• Varnish 2.1.3 or newer

Due to issues with unicode in Python2’s httplib library, we recommend installing the pycurl package. Disco will transparently use pycurl when available.

Install Disco¶

git clone git://github.com/discoproject/disco.git $DISCO_HOME
cd $DISCO_HOME
make
cd lib && python setup.py install --user && cd ..
bin/disco nodaemon

make

make install

cd lib
python setup.py install --user
cd ..

bin/disco nodaemon

bin/disco start

ps aux | grep beam.*disco

Configure Authentication¶

Next we need to enable passwordless login via ssh to all servers in the Disco cluster. If you have only one machine, you need to enable passwordless login to localhost for the Disco user.

Run the following command as the Disco user, assuming that it doesn’t have valid ssh-keys already:

ssh-keygen -N '' -f ~/.ssh/id_dsa

If you have one server (or shared home directories), say:

cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

Otherwise, repeat the following command for all the servers nodeX in the cluster:

ssh-copy-id nodeX

Now try to login to all servers in the cluster or localhost, if you have only one machine. You should not need to give a password nor answer to any questions after the first login attempt.

As the last step, if you run Disco on many machines, you need to make sure that all servers in the Disco cluster use the same Erlang cookie, which is used for authentication between Erlang nodes. Run the following command as the Disco user on the master server:

scp ~/.erlang.cookie nodeX:

Repeat the command for all the servers nodeX.

Add nodes to Disco¶

At this point you should have Disco up and running. The final step, before testing the system, is to specify which servers are available for Disco. This is done via Disco’s web interface.

Point your browser at http://<DISCO_MASTER_HOST>:<DISCO_PORT>, where DISCO_MASTER_HOST and DISCO_PORT should be replaced with their actual values. Normally you can use http://localhost:8989, if you run Disco locally or through an SSH tunnel.

You should see the Disco main screen (see a screenshot). Click configure on the right side of the page. On the configuration page, click add row to add a new set of available nodes. Click the cells on the new empty row, and add hostname of an available server (or a range of hostnames) in the left cell, and the number of available cores (CPUs) on that server in the right cell. Once you have entered a value, click the cell again to save it.

You can add as many rows as needed to fully specify your cluster, which may have varying number of cores on different nodes. Click save table when you are done.

Add the localhost¶

If you have only a single machine, the resulting table should look like this, assuming that you have two cores available for Disco:

Warning

It is not advised to use the master as a slave node in a serious Disco cluster.

Add multiple nodes in the same line¶

You can also specify multiple nodes on a single line, if the nodes are named with a common prefix, as here:

This table specifies that there are 30 nodes available in the cluster, from nx01 to nx30 and each node has 8 cores.

Test the System¶

Now Disco should be ready for use.

We can use the following simple Disco script that computes word frequencies in a text file to see that the system works correctly.

from disco.core import Job, result_iterator

def map(line, params):
    for word in line.split():
        yield word, 1

def reduce(iter, params):
    from disco.util import kvgroup
    for word, counts in kvgroup(sorted(iter)):
        yield word, sum(counts)

if __name__ == '__main__':
    job = Job().run(input=["http://discoproject.org/media/text/chekhov.txt"],
                    map=map,
                    reduce=reduce)
    for word, count in result_iterator(job.wait(show=True)):
        print(word, count)

Run the script as follows from DISCO_HOME:

python examples/util/count_words.py

Disco attempts to use the current hostname as DISCO_MASTER_HOST, if it is not defined in any settings file.

If you are runing Disco on multiple machines you must use the same version of Python for running Disco scripts as you use on the server side.

You can run the script on any machine that can access the master. The safest bet is to test the script on the master node itself.

If the machine where you run the script can access the master node but not other nodes in the cluster, you need to set DISCO_PROXY. The proxy address should be the same as the master’s above. This makes Disco fetch results through the master node, instead of connecting to the nodes directly.

If the script produces some results, congratulations, you have a working Disco setup! If you are new to Disco, you might want to read the Tutorial next.

If the script fails, see the section about Troubleshooting Disco installation.

Install From Source¶

Note: On FreeBSD, replace all of the instances of make with gmake.

Assuming you have already gotten Disco running out of the source directory, as described in Install Disco, to install system-wide, just run make install as root:

make install

This will build and install the Disco master to your system (see the Makefile for exact directory locations). You can specify DESTDIR and prefix, in compliance with GNU make.

On systems that are intended to function as Disco worker nodes only, you can use the make install-node target instead.

System Settings¶

make install installs a configuration file to /etc/disco/settings.py that is tuned for clusters, not a single machine.

By default, the settings assume that you have at least three nodes in your cluster, so DDFS can use three-way replication. If you have fewer nodes, you need to lower the number of replicas in /etc/disco/settings.py:

DDFS_TAG_MIN_REPLICAS=1
DDFS_TAG_REPLICAS=1
DDFS_BLOB_REPLICAS=1

Most likely you do not need to modify anything else in this file right now, but you can change the settings here, if the defaults are not suitable for your system.

See disco.settings for more information.

Creating a disco user¶

You can use any account for running Disco, however it may be convenient to create a separate disco user. Among other advantages, this allows setting resource utilization limits for the disco user (through limits.conf or similar mechanism).

Since Disco places no special requirements on the user, (except access to certain ports and the ability to execute and read its files), simply follow the guidelines of your system when it comes to creating new users.

Keeping Disco Running¶

You can easily integrate disco into your system’s startup sequence. As an example, you can see how disco-master.init is implemented in Disco’s debian packaging.

Configuring DDFS Storage¶

On the Disco nodes, DDFS creates by default a subdirectory named vol0 under the DDFS_DATA directory to use for storage. If you have one or more dedicated disks or storage areas you wish to use instead, you can mount them under the directory specified by DDFS_DATA as subdirectories named vol0, vol1 and so on.

1. Prepare input data¶

Disco can distribute computation only as well as data can be distributed. In general, we can push data to Disco Distributed Filesystem, which will take care of distributing and replicating it.

Lets chunk and push the data to a tag data:bigtxt:

ddfs chunk data:bigtxt ./bigfile.txt

We should have seen some output telling us that the chunk(s) have been created. We can also check where they are located:

ddfs blobs data:bigtxt

and make sure they contain what you think they do:

ddfs xcat data:bigtxt | less

If you used the file provided above, you should have only ended up with a single chunk. This is because the default chunk size is 64MB (compressed), and the bigfile.txt is only 12MB (uncompressed). You can try with a larger file to see that chunks are created as needed.

2. Write job functions¶

Next we need to write map and reduce functions to count words. Start your favorite text editor and create a file called count_words.py. First, let’s write our map function:

def fun_map(line, params):
        for word in line.split():
                yield word, 1

Quite compact, eh? The map function takes two parameters, here they are called line and params. The first parameter contains an input entry, which is by default a line of text. An input entry can be anything though, since you can define a custom function that parses an input stream (see the parameter map_reader in the Classic Worker). The second parameter, params, can be any object that you specify, in case that you need some additional input for your functions.

For our example, we can happily process input line by line. The map function needs to return an iterator over of key-value pairs. Here we split a line into tokens using the builtin string.split(). Each token is output separately as a key, together with the value 1.

Now, let’s write the corresponding reduce function:

def fun_reduce(iter, params):
        from disco.util import kvgroup
        for word, counts in kvgroup(sorted(iter)):
                yield word, sum(counts)

The first parameter, iter, is an iterator over those keys and values produced by the map function, which belong to this reduce instance (see partitioning).

In this case, words are randomly assigned to different reduce instances. Again, this is something that can be changed (see partition() for more information). However, as long as all occurrences of the same word go to the same reduce, we can be sure that the final counts are correct.

The second parameter params is the same as in the map function.

We simply use disco.util.kvgroup() to pull out each word along with its counts, and sum the counts together, yielding the result. That’s it. Now we have written map and reduce functions for counting words in parallel.

3. Run the job¶

Now the only thing missing is a command for running the job. There’s a large number of parameters that you can use to specify your job, but only three of them are required for a simple job like ours.

In addition to starting the job, we want to print out the results as well. First, however, we have to wait until the job has finished. This is done with the wait() call, which returns results of the job once has it has finished. For convenience, the wait() method, as well as other methods related to a job, can be called through the Job object.

A function called result_iterator() takes a list of addresses to the result files, that is returned by wait(), and iterates through all key-value pairs in the results.

The following example from examples/util/count_words.py runs the job, and prints out the results:

from disco.core import Job, result_iterator

def map(line, params):
    for word in line.split():
        yield word, 1

def reduce(iter, params):
    from disco.util import kvgroup
    for word, counts in kvgroup(sorted(iter)):
        yield word, sum(counts)

if __name__ == '__main__':
    job = Job().run(input=["http://discoproject.org/media/text/chekhov.txt"],
                    map=map,
                    reduce=reduce)
    for word, count in result_iterator(job.wait(show=True)):
        print(word, count)

Now comes the moment of truth.

Run the script as follows:

python count_words.py

If everything goes well, you will see that the job executes. The inputs are read from the tag data:bigtxt, which was created earlier. Finally the output is printed. While the job is running, you can point your web browser at http://localhost:8989 (or some other port where you run the Disco master) which lets you follow the progress of your job in real-time.

You can also set DISCO_EVENTS to see job events from your console:

DISCO_EVENTS=1 python count_words.py

In this case, the events were anyway printed to the console, since we specified show=True.

What next?¶

As you saw, creating a new Disco job is pretty straightforward. You could extend this simple example in any number of ways. For instance, by using the params object to include a list of stop words.

You could continue on with Extended Tutorial which is intended as a follow-on tutorial to this one.

If you pushed the data to Disco Distributed Filesystem, you could try changing the input to tag://data:bigtxt, and add map_reader = disco.worker.task_io.chain_reader.

You could follow the DiscoDB Tutorial, to learn more about using discodb with Disco.

You could try using sum_combiner(), to make the job more efficient.

You can also experiment with custom partitioning and reader functions. They are written in the same way as map and reduce functions. Just see some examples in the disco.worker.classic.func module. After that, you could try chaining jobs together, so that output of the previous job becomes input for the next one.

The best way to learn is to pick a problem or algorithm that you know well, and implement it with Disco. After all, Disco was designed to be as simple as possible so you can concentrate on your own problems, not on the framework.

1. Background and sample input¶

Let’s first prepare a sample input data set that’s small enough and simple enough for us to follow and know what to expect on output. We will prepare two sets of input in csv format to be “joined” together using the first entry in each row as the key to match (join) on. Create a file named set_A.csv containing the following text:

1,"alpha"
2,"beta"
3,"gamma"
4,"delta"
5,"epsilon"

Create a second file named set_B.csv containing the following text:

1,"who"
2,"what"
3,"where"
4,"when"
5,"why"
6,"how"

When we inner_join these two datasets using the first entry in each row as its key, we would like to see output that looks something like this:

1,"alpha","who"
2,"beta","what"
3,"gamma","where"
4,"delta","when"
5,"epsilon","why"

Note that there is no line in the output for key=6 as seen in the input data of set_B.csv because it did not have a matched pair for that key in set_A.csv. Please also note that we would expect the output to be the same even if the order of the lines were scrambled in either of the two input data sets.

You should now have two files in your working directory named set_A.csv and set_B.csv which contain 5 and 6 lines, respectively, of text data.

2. Split input data into chunks¶

In the introductory Tutorial, we made use of a DDFS (Disco Distributed Filesystem) command, ddfs chunk, to split input data into chunks and copy it onto DDFS. To provide a more concrete sense of how to chunk input data, let’s instead split our input data before we push it to DDFS. When we do push our already-split data to DDFS, we will tell DDFS to treat the distinct chunks as one.

As alluded to before, there are many strategies for performing efficient join operations inside MapReduce frameworks. Here we will take the approach of combining our two input data sets (A and B) into a single input stream. With a single input stream, it’s easier to see how to split up the input, do work on it, then merge it back together. This approach doesn’t necessarily harm performance but there are different strategies tuned for optimal performance depending upon the nature of your data. (Search the net for “mapreduce join” to see the wealth of competing strategies out there.)

Assuming a unix-like environment from here on, start by combining our two input files:

% cat set_A.csv set_B.csv > both_sets.csv

Next, we want to split our both_sets.csv file into chunks with 2 lines each. You can do this with a text editor yourself, by hand, or we can make use of the convenient unix utility split to do the job for us:

% split -l 2 both_sets.csv

Running split as above should create 6 files named xaa through xaf. You can quickly verify this by performing a count of the lines in each file and seeing that it adds up to 11:

% wc -l xa?
  2 xaa
  2 xab
  2 xac
  2 xad
  2 xae
  1 xaf
 11 total

Now that we’ve split the input data ourselves into 6 chunks, let’s push our split data into DDFS and label it all with a single tag, data:both_sets, so that we can refer to all our chunks as one:

% ddfs push data:both_sets ./xa?

You can verify that all 11 lines made it into DDFS and are accessible via that single tag by asking to cat it back to the screen:

% ddfs cat data:both_sets

By splitting our input data into 6 chunks, we are now set up to perform 6 executions of our map function (which we have yet to implement). If you have a processor with 6 cores, you could conceivably perform all 6 map operations in parallel at the same time. If you have more than 6 cores either on one processor or across multiple processors available to Disco, you’ll only be able to make use of, at most, 6 of them at one time during the map phase of a MapReduce job. In general: If you want more map operations to be running at the same time, make more chunks (smaller chunks). Taking it too far, if you make more chunks than you have cores, you won’t get further speedup from parallelism.

You should now have the 11 lines of input csv-format data stored in DDFS in 6 chunks under the tag data:both_sets. While not necessarily the best approach for splitting and importing your largest datasets into DDFS, it may prove helpful to remember that you can chunk your data all at once or bring it in in pieces.

You can also set a limit for the size of the chunks to increase the number of the chunks for a fixed size file. The default maximum chunk size is 64 MB. You can use:

% ddfs chunk -S 0.1 data:other_sets ./both_sets.csv

This will result in each chunk being smaller than 0.1 * 1MB. For this small file, only one chunk will be created.

3. Write a job using a derived class¶

In the introductory Tutorial, we defined a map function and a reduce function, and then supplied them as parameters to Job().run(). But there’s more fun to be had by deriving a new class from Job. Let’s start by declaring our new class and saving it in a source file named simple_innerjoin.py:

class CsvInnerJoiner(Job):
    def map(self, row, params):
        # TODO
        pass

    def reduce(self, rows_iter, out, params):
        # TODO
        pass

Before we turn attention to implementing either of the map or reduce methods, we should consider our need, in this example, to read input that’s in csv format. A convenient solution is to implement map_reader() in our class:

@staticmethod
def map_reader(fd, size, url, params):
    reader = csv.reader(fd, delimiter=',')
    for row in reader:
        yield row

This will allow us to implement map() to operate on one row’s worth of input data at a time without needing to worry about raw input format.

Our strategy with our map and reduce methods will be to first sort all of the input data by their unique keys (which will put row 4 from set_A.csv right next to / in front of row 4 from set_B.csv), then merge consecutive rows having the same unique key. This puts most of the burden on our reduce() implementation, but we’ll ease that a bit in a later pass. Since map() does not need to do much other than serve as a pass-through (quickly), modify our placeholder for map() to read:

def map(self, row, params):
    yield row[0], row[1:]

This will separate the unique key (in position 0) from all the other data on a row (assuming we want to re-use this for something more interesting than our fairly trivial input data set so far).

Now we ask reduce() to do the real work in its updated definition:

def reduce(self, rows_iter, out, params):
    from disco.util import kvgroup
    from itertools import chain
    for url_key, descriptors in kvgroup(sorted(rows_iter)):
        merged_descriptors = list(chain.from_iterable(descriptors))
        if len(merged_descriptors) > 1:
            out.add(url_key, merged_descriptors)

Again, as in Tutorial, we are using disco.util.kvgroup() to group together consecutive rows in our sorted input and hand them back as a group (iterable). Note our test to see if we have a matched pair or not is somewhat fragile and may not work for more general cases – we highlight this as an area for improvement for the reader to consider later.

Let’s round out our simple_innerjoin.py tool by making it easy to supply names for input and output, while also making our output come out in csv format – adding to the bottom of simple_innerjoin.py:

if __name__ == '__main__':
    input_filename = "input.csv"
    output_filename = "output.csv"
    if len(sys.argv) > 1:
        input_filename = sys.argv[1]
        if len(sys.argv) > 2:
            output_filename = sys.argv[2]

    from simple_innerjoin import CsvInnerJoiner
    job = CsvInnerJoiner().run(input=[input_filename])

    with open(output_filename, 'w') as fp:
        writer = csv.writer(fp)
        for url_key, descriptors in result_iterator(job.wait(show=True)):
            writer.writerow([url_key] + descriptors)

In the prior Tutorial, all output flowed to the screen (stdout) but here we capture the output flowing from our job into a file in csv format. We chose to use the csv format throughout this Extended Tutorial for convenience but clearly other methods of redirecting output and formatting it to your own needs are possible in the same way.

4. Results and exploring partitions¶

We should now be set up to run our job with 6 input chunks corresponding to 6 invocations of our map() method and the output of those map runs will flow into 1 invocation of our reduce() method to then produce our final csv result file. Launching from the command-line:

% python simple_innerjoin.py data:both_sets output.csv

At this point, please check that the output found in the file output.csv matches what was expected. (Pedants can play further with formatting and quotation rules via the csv module, to taste.) If you instead encounter errors, please double-check that your file faithfully matches the code outlined thus far and please double-check that you can still run the example from the introductory Tutorial.

Thus far we’ve been running parallel invocations of map() but not of reduce() – let’s change that by requesting that the output from the map phase be divided into 2 partitions. Add the following line to the very top of our definition of the CsvInnerJoiner class, to look something like this:

class CsvInnerJoiner(Job):
    partitions = 2

    ...*truncated*...

Run the job again from the command-line and this time you may find that while the output might be correct, the output is no longer in sort-order. This is because we did not sort over all rows – only the rows handed to a particular invocation of reduce() were sorted, though we still get to see the output from parallel invocations of reduce() concatenated together in our single output csv file.

This helps highlight a problem we’re going to have once we start throwing larger volumes of data at this Disco job: invoking sorted() requires a potentially large amount of memory. Thankfully Disco provides, as part of its framework, an easier solution to this common need for working with sorted results in the reduce step. At the top of our definition of the CsvInnerJoiner class, let’s add the following line:

class CsvInnerJoiner(Job):
    partitions = 2
    sort = True

    ...*truncated*...

Simultaneously, we can remove the use of sorted() from the one line in our implementation of reduce() so that it now reads as:

def reduce(self, rows_iter, out, params):
    from disco.util import kvgroup
    from itertools import chain
    for url_key, descriptors in kvgroup(rows_iter):
        merged_descriptors = list(chain.from_iterable(descriptors))
        if len(merged_descriptors) > 1:
            out.add(url_key, merged_descriptors)

Now the work of sorting the results flowing from the mappers is done for us by the framework and that sort is performed across all mappers’ results before being partitioned and handed as input to the reducers.

5. Big(ger) Data¶

Let’s quickly generate a bigger input data set with which to work. The following one-liner can be modified to generate as little or as much sample data as you have patience / disk space to hold – modify the 1000000 near the end of the line to create as many rows of data as you like:

% python -c "import csv, sys, random; w = csv.writer(sys.stdout);
[w.writerow([i, int(999999*random.random())]) for i in range(1000000)]" > input1.csv

Run it twice (saving the first run’s output in a different name from the second run’s) to give yourself two sets of input data just as before. Then follow the steps from either this Extended Tutorial or the prior introductory Tutorial to chunk the input data and push it to DDFS in whatever manner you like. (Let’s assume you tag your chunked input data as data:bigger_sets in DDFS.)

The only modification to simple_innerjoin.py that we suggest, depending upon how large your newly generated input data set is, is to increase the number of partitions to ratchet up the number of parallel runs of reduce(). Then go ahead and run your job in the same way:

% python simple_innerjoin.py data:bigger_sets bigger_output.csv

By monitoring the processes on the system(s) where you’ve configured Disco, you will hopefully be able to observe individual workers performing their map tasks and reduce tasks, the framework doing your sorting work for you in between, and how much cpu processing time is being used versus time spent waiting on disk or other resources. Having a larger dataset with a longer runtime makes observing these things much easier.

Note that you may quickly find your disk access speed to become a bottleneck and for this reason and others you should consider playing with the number of partitions as well as the number of input chunks (how many reducers and mappers, respectively) to find your system’s optimal throughput for this job.

As a variation on the above, remember that our simple_innerjoin.py script has the capability to read its input data from a local file instead of DDFS – try running again with a local file supplied as the location of the input (instead of data:bigger_sets). Did you get an error message with “Invalid tag (403)”? If so, you need to ensure Disco recognizes that you are supplying a filename and not the name of a tag. Did you get an error message with “IOError: [Errno 2] No such file or directory”? If so, you either need to supply the full path to the file (not a relative path name) or that path may not be available to Disco everywhere (if so, a good reason to use DDFS again). Was your run faster or slower than using DDFS?

After playing with ever larger volumes of data and tweaking the controls that Disco provides, you’ll quickly gain confidence in being able to throw any size job at Disco and knowing how to go about implementing a solution.

simple_innerjoin.py listing¶

Complete source all in one place:

from disco.core import Job, result_iterator
import csv, sys


class CsvInnerJoiner(Job):
    partitions = 2
    sort = True

    def map(self, row, params):
        yield row[0], row[1:]

    @staticmethod
    def map_reader(fd, size, url, params):
        reader = csv.reader(fd, delimiter=',')
        for row in reader:
            yield row

    def reduce(self, rows_iter, out, params):
        from disco.util import kvgroup
        from itertools import chain
        #for url_key, descriptors in kvgroup(sorted(rows_iter)):
        for url_key, descriptors in kvgroup(rows_iter):
            merged_descriptors = list(chain.from_iterable(descriptors))
            if len(merged_descriptors) > 1:
                out.add(url_key, merged_descriptors)


if __name__ == '__main__':
    input_filename = "input.csv"
    output_filename = "output.csv"
    if len(sys.argv) > 1:
        input_filename = sys.argv[1]
        if len(sys.argv) > 2:
            output_filename = sys.argv[2]

    from simple_innerjoin import CsvInnerJoiner
    job = CsvInnerJoiner().run(input=[input_filename])

    with open(output_filename, 'w') as fp:
        writer = csv.writer(fp)
        for url_key, descriptors in result_iterator(job.wait(show=True)):
            writer.writerow([url_key] + descriptors)

What next?¶

A natural next step in experimenting with partitioning involves chaining jobs together since the number of partitioned outputs from one job becomes the number of chunked inputs for the next. As a baby step, you could move the reduce() method implemented above into a second, chained job and replace it in the first job with a do-nothing substitute like disco.worker.classic.func.nop_reduce().

As already mentioned in the introductory Tutorial, the best way to learn is to pick a problem or algorithm that you know well, and implement it with Disco. After all, Disco was designed to be as simple as possible so you can concentrate on your own problems, not on the framework.

Make sure Disco is not running¶

If you have started Disco earlier, try to stop the master using disco stop (or C-c if you are running with disco nodaemon). If you cannot seem to stop Disco this way, kill the beam processes by hand,

ps aux | grep beam.*disco

kill PID

Is the master starting?¶

Start Disco by saying:

disco nodaemon

If everything goes well, you should see a bunch of =INFO REPORT= messages printed to the screen. If you see any =ERROR REPORT= messages, something is wrong, and you should try to resolve the particular issue Erlang is reporting. These messages often reveal what went wrong during the startup.

If you see something like this:

application: disco
exited: {bad_return,{{disco_main,start,[normal,[]]},
        {'EXIT',["Specify ",scgi_port]}}}

Disco is trying to start up properly, but your Erlang installation probably doesn’t work correctly and you should try to re-install it.

disco -v

If the master is running, you can proceed to the next step (you can double check with ps as in Make sure Disco is not running). If not, the master didn’t start up properly.

Are there any nodes on the status page?¶

Now that we know that the master process is running, we should be able to configure the system. Open your web browser and go to http://localhost:8989/ (or whatever your DISCO_MASTER_HOST and DISCO_PORT are set to). The Disco status page should open.

Do you see any boxes with black title bars on the status page (like in this screenshot)? If not, add nodes to the system as instructed in Add nodes to Disco.

If adding nodes through the web interface fails, you can try editing the config file manually. For instance, if you replace DISCO_ROOT in the following command, it will create a configuration file with one node:

echo '[["localhost", "1"]]' > DISCO_ROOT/disco_4441.config

Now is a good time to try to run a Disco job. Go ahead and retry the installation test. You should see the job appear on the Disco status page. If the job succeeds, it should appear with a green box on the job list. If it turns up red, we need to continue debugging.

Are slaves running?¶

In addition to the master process on the master node, Erlang runs a slave on each node in a Disco cluster.

Make sure that the slave is running:

ps aux | grep -o disco.*slave@

If is is running, you should see something like this:

disco_8989_master@discodev -sname disco_8989_slave@
disco.*slave@

If you get a similar output, go to Do workers run?. If not, read on.

ssh localhost erl

[user@somehost dir]$ disco debug
Erlang VERSION

Eshell VERSION (abort with ^G)
(testmaster@somehost)1> slave:start(localhost, "testnode").
{ok,testnode@localhost}
(testmaster@somehost)1> net_adm:ping(testnode@localhost).
pong

Is your firewall configured correctly?¶

Disco requires a number of ports to be accessible to function properly.

• 22 - SSH
• 8990 - DDFS web API
• 8989 - Disco web interface/API. Must be unblocked on slaves and the master.
• 4369 - Erlang port mapper
• 30000 to 65535 - Communication between Erlang slaves

Is your DNS configured correctly?¶

Disco uses short DNS names of cluster nodes in its configuration. Please ensure that short hostnames were entered in the Add nodes to Disco step, and that DNS resolves these short names correctly across all nodes in the cluster.

Do workers run?¶

The master is responsible for starting individual processes that execute the actual map and reduce tasks. Assuming that the master is running correctly, the problem might be in the worker.

See what happens with the following command:

ssh localhost "python DISCO_HOME/lib/disco/worker/classic/worker.py"

Where DISCO_HOME in this case must be the Disco source directory. It should start and send a message like this:

WORKER 32 {"version": "1.0", "pid": 13492}

If you get something else, you may have a problem with your PATH or Python installation.

Still no success?¶

If the problem persists, or you can’t get one of the steps above working, do not despair! Report your problem to friendly Disco developers on IRC or the mailing list. Please mention in your report the steps you followed and the results you got.

Develop Disco¶

Get Disco from github. You can easily fork a repository of your own (just click “fork”). You can set up a development environment on a single machine or a cluster.

Mailing list / discussion groups¶

We have a Google group for Disco which serves as our mailing list. Subscribe here.

IRC¶

Join Disco discussions at our IRC channel #discoproject at Freenode. This is usually the fastest way to get answers to your questions and get in touch with Disco developers.

If you haven’t used IRC before, see here how to get started.

Roadmap / Wiki¶

We are using the issues list as a roadmap for Disco. You can report bugs and wishlist features for Disco there. Also see the wiki for other miscellaneous information.

Monitoring the cluster¶

An overall view of the state of the cluster is provided on the status page, which is the default page for the web user interface, and is accessible by clicking status on the top right of any page. This shows a node status box for each node in the cluster.

A black title background in the top row of the status box indicates that the node is connected to the master, whereas a node that the master is unable to connect to will have a crimson title background. A blacklisted node will have a lavender background for the entire status box. Note that the master still attempts to connect to blacklisted nodes, which means that the blacklist status and the connection status for a node are independent.

The second row in the status box shows a box for each of the configured cores on the node, and shows the busy cores in yellow. The third row shows amount of available disk space in the DDFS filesystem on the node.

The fourth row shows some job statistics for the node. The leftmost number shows the number of tasks that successfully completed on the node. The middle number indicates the number of tasks that were restarted due to them experiencing recoverable failures. The rightmost number indicates the number of tasks that crashed due to non-recoverable errors. Since a crashing task also causes its job to fail, this number also indicates the number of jobs that failed due to tasks crashing on this node.

The job listing along the top right side of the page shows running jobs in yellow, completed jobs in green, and crashed or killed jobs in pink. Hovering over a running job highlights the cores that the tasks of the job are using on each node in light blue.

Below the job listing on the right is some information from the last DDFS garbage collection run. A table mentions the number of tags and blobs that were kept after the last GC run and their total sizes in bytes, along with similar information for deleted blobs and tags.

Using a proxy¶

A cluster is sometimes configured with a few head nodes visible to its users, but most of the worker nodes hidden behind a firewall. Disco can be configured to work in this environment using a HTTP proxy, provided the node running the Disco master is configured as one of the head nodes, and the port used by the proxy is accessible to the Disco users.

To use this mode, the Disco settings on the master (in the /etc/disco/settings.py file) should set DISCO_PROXY_ENABLED to "True". Disco then starts a HTTP proxy specified by DISCO_HTTPD (defaulting to lighttpd) on port DISCO_PROXY_PORT (defaulting to 8999) with a configuration that proxies HTTP access to the Disco worker nodes. Currently, Disco supports generating configurations for the lighttpd and varnish proxies. The Disco master needs to be restarted for any changes in these settings to take effect.

To use this proxy from the client side, i.e. in order to use the disco and ddfs commands, the users need to set DISCO_PROXY in their environment, or in their local /etc/disco/settings.py file. The value of this should be in the format http://<proxy-host>:<proxy-port>, with <proxy-host> and <proxy-port> specifying how the proxy is accessible to the user.

Blacklisting a Disco node¶

You may decide that you need to reduce the load on a node if it is not performing well or for some other reason. In this case, you can blacklist a node, which informs Disco that the node should not be used for running any tasks or storing any new DDFS data. However, Disco will still use the node for reading any DDFS data already stored on that node. Note that blacklisting a node will not trigger re-replication of data away from the node (however, see below on blacklisting a DDFS node).

Blacklisting a node is done on the configuration page, accessible by clicking configure on the top right side of any page. Type the name of the node in the text entry box under Blacklisted nodes for Disco, and press enter. The node should now show up above the text entry box in the list of blacklisted nodes. Any node in this list can be clicked to whitelist it.

The set of blacklisted nodes is persisted, so that it is not lost across a restart.

Adjusting Disco system settings¶

Disco provides a tunable parameter max_failure_rate to administrators to control the number of recoverable task failures a job can experience before Disco fails the entire job. This parameter can be set on the configuration page.

Removing nodes from Disco¶

You probably went through the process of adding a node to the cluster when you configured the Disco cluster in Add nodes to Disco. Removing a node from the Disco configuration is a very similar process, except that you need to click remove next to the node(s) you wish to remove, and then click save table when you are done.

Blacklisting a DDFS node¶

There are various ways of removing a node that hosts DDFS data from the Disco cluster, with differing implications for safety and data availability.

• The node could be physically removed from the cluster, but left in the Disco configuration. In this case, it counts as a failed node, which reduces by one the number of additional node failures that could be safely tolerated. Garbage collection and Re-replication will attempt the replication of any missing live blobs and tags.
• The node could be physically removed from the cluster as well as the Disco configuration. In this case, Disco treats the missing node as an unknown node instead of as a failed node, and the number of additional node failures tolerated by DDFS does not change. However, this voids safety, since Disco might allow more nodes hosting DDFS data to fail than is safe. For example, if the replication factor is set to 3, Disco might treat two additional node failures as safe, whereas actually those two nodes might be hosting the last two remaining replicas of a blob, the third replica being lost when the first node was removed from the configuration. If this happens, Garbage collection and Re-replication will not suffice to replicate the missing blobs and tags, and some data might be permanently lost.

The drawback of both of these approaches is that there is no indication provided by Disco as to when, if ever, DDFS is in a consistent state again with respect to the data that was hosted on the removed node.

DDFS now allows scheduling the removal of a node from DDFS, by putting the node on a DDFS blacklist, which is specified using the text entry box labeled Blacklisted nodes for DDFS. This makes Garbage collection and Re-replication actively replicate data away from that node; that is, additional replicas are created to replace the blobs hosted on a blacklisted node, and when safe, references to the blobs on that node are removed from any referring tags. DDFS data on the blacklisted node is however not deleted.

In addition, DDFS now provides an indication when all the data and metadata that was hosted on that node has been re-replicated on other cluster nodes, so that that node can be safely removed from the Disco cluster (both physically, as well as from the configuration) with a guarantee that no data has been lost. The indication is provided by the node entry being highlighted in green in the blacklist. It may require several runs of Garbage collection and Re-replication to re-replicate data away from a node; since by default it runs once a day, several days may be needed before a DDFS blacklisted node becomes safe for removal.

Handling a master failure¶

Disco is currently a single master system, which means it has a single point of failure. This master controls both job scheduling as well as the Disco Distributed Filesystem. The failure of the master will result in the termination of currently running jobs and loss of access to the Disco Distributed Filesystem. However, it will not result in any loss of data in DDFS, since all metadata in DDFS is replicated, just like data. The only centralized static information is the Disco settings file on the master (specified by the DISCO_SETTINGS_FILE, which defaults to /etc/disco/settings.py for installation), and the Disco cluster configuration, maintained in the file specified by the DISCO_MASTER_CONFIG setting. You can examine all the settings for Disco using the disco -v command.

A failed Disco master can be replaced by installing the Disco master on a new machine (or even an existing Disco node, though this is not recommended for large or busy clusters). See Installing Disco System-Wide for details on installing the Disco master. On the replacement machine, you will need to copy the settings and configuration files from the original master into their expected locations. For this reason, it is a good idea to backup these files after any change. The DISCO_SETTINGS_FILE is manually modified, while the DISCO_MASTER_CONFIG file is managed by the Disco master. The config file is changed whenever nodes are added or removed as members of the cluster, or any blacklist.

Map Flows¶

The two basic modes of operation for the map phase correspond directly to writing either partitioned or non-partitioned output.

Partitioned Map¶

For partitioned map, each output is written to one of the partitions:

Non-Partitioned Map¶

For non-partitioned map, every map task creates a single output. In other words, the input-output relationship is exactly 1 to 1:

Single-Partition Map¶

Notice that for partitioned output with N partitions, exactly N files will be created for each node, regardless of the number of maps. If the map tasks run on K nodes, exactly K * N files will be created. Whereas for non-partitioned output with M inputs, exactly M output files will be created.

This is an important difference between partitioned output with 1 partition, and non-partitioned output:

The default number of partitions for map is 1. This means that by default if you run M maps on K nodes, you end up with K files containing the results. In older versions of Disco, there were no partitions by default, so that jobs with a huge number of inputs produced a huge number of outputs. If M >> K, this is suboptimal for the reduce phase.

Reduce Flows¶

The basic modes of operation for the reduce phase correspond to partitioned/non-partitioned input (instead of output as in the map phase).

Normal Partitioned Reduce Flow¶

As you might expect, the default is to distribute the reduce phase.

Non-Partitioned Reduce¶

For non-partitioned input, there can only ever be 1 reduce task:

The situation is slightly more complicated for partitioned input, as there is a choice to be made whether or not to merge the partitions, so that all results are handled by a single reduce.

Introduction¶

Disco Distributed Filesystem (DDFS) provides a distributed storage layer for Disco. DDFS is designed specifically to support use cases that are typical for Disco and mapreduce in general: Storage and processing of massive amounts of immutable data. This makes it very suitable for storing, for instance: log data, large binary objects (photos, videos, indices), or incrementally collected raw data such as web crawls.

In this sense, DDFS is complementary to traditional relational databases or distributed key-value stores, which often have difficulties in scaling to tera- or petabytes of bulk data. Although DDFS stands for Disco Distributed filesystem, it is not a general-purpose POSIX-compatible filesystem. Rather, it is a special purpose storage layer similar to the Google filesystem or related open-source projects such as Hadoop Distributed Filesystem (HDFS), MogileFS or Tabled.

DDFS is a low-level component in the Disco stack, taking care of data distribution, replication, persistence, addressing and access. It does not provide a sophisticated query facility in itself but it is tightly integrated with Disco jobs. Disco can store job results to DDFS, providing persistence for and easy access to processed data.

DDFS is a tag-based filesystem: Instead of having to organize data to directory hierarchies, you can tag sets of objects with arbitrary names and retrieve them later based on the given tags. For instance, tags can be used to timestamp different versions of data, or denote the source or owner of data. Tags can contain links to other tags, and data can be referred to by multiple tags; tags hence form a network or a directed graph of metadata. This provides a flexible way to manage terabytes of data assets. DDFS also provides a mechanism to store arbitrary attributes with the tags, for instance, to denote data type.

DDFS is schema-free, so you can use it to store arbitrary, non-normalized data. However, it is not suitable for storing data items that are very small (fewer than 4K) or that need to be updated often, such as user passwords or status indicators. You can store frequently changing data in a key-value store or a relational database. If you need to analyze this data with Disco, you can dump a snapshot of the full database to DDFS, for instance, to update your user models every night.

DDFS is horizontally scalable. New nodes can be added to the storage cluster on the fly, using the Disco web UI. All heavy IO on bulk data is distributed, so there are no bottlenecks limiting the amount of data that DDFS can handle. Only metadata is handled centrally, ensuring that it is kept consistent all the time.

DDFS is designed to operate on commodity hardware. Fault-tolerance and high availability are ensured by K-way replication of both data and metadata, so the system tolerates K-1 simultaneous hardware failures without interruptions. DDFS stores data and metadata on normal local filesystems, such as ext3 or xfs, so even under a catastrophic failure data is recoverable using standard tools.

Concepts¶

DDFS operates on two concepts: Blobs and Tags.

Overview¶

Consider that you have a log file containing data of a single day.

For DDFS, this is a blob. When you push the blob to DDFS using DDFS APIs, a DDFS client distributes the blob to K nodes.

By default, K is 3, so you get three identical replicas of the blob. DDFS and Disco can utilize any of the replicas, in case some of them are unavailable due to disk or server failure. DDFS ensures that you will always have K replicas, even if disks fail, by re-replicating blobs if needed. This guarantees that your data is truly persistent.

Even persistent data is not very valuable if it cannot be accessed easily. The blobs distributed above are stored on three random nodes. To be able to use them efficiently, metadata storing addresses of the blobs is needed. DDFS uses tags for this purpose.

The green tag allows you to query data behind data:log:website using DDFS APIs and retrieve a tag object that contains URLs to the blobs. You can access the blobs using their URLs over HTTP as usual, or give the list to Disco to be used as inputs for a Map/Reduce job. Naturally metadata should not be lost under any circumstances, so tags are replicated and distributed to many nodes similarly to blobs.

Each blob must have at least one tag linking to it. Otherwise the blob is practically unaccessible or orphaned. Orphaned blobs are eventually deleted by the garbage collector. Correspondingly, if you want to delete a set of blobs from DDFS, you must delete all links (or tags) referencing the blobs which makes them orphaned and subject to eventual removal.

Eventually you want to add more daily logs (blobs) under the tag data:log:website. Each daily log is replicated separately, so the tag ends up containing many replication sets, that is, lists of URLs that pointing at replicas of a blob. Replications sets are represented by dotted boxes above.

DDFS allows tags to reference other tags. This is a very powerful feature which makes it possible to implement tag hierarchies and graphs. For instance, the tag user:mike above links to all tags owned by Mike. DDFS APIs provides functions to traverse the tag graph, so it is straightforward to retrieve all tags and blobs owned by Mike.

Tags may also reference overlapping sets of blobs, as in data:log:peakday above. This feature is useful if you want to provide many alternative views to the same data. DDFS is designed to scale to millions of tags, so you can use them without hesitation.

Tags also support a token-based authorization mechanism to control read and write access. If a write-token is specified for a tag, all operations that wish to modify the tag will need to provide this write-token. Without this token, any write operation will return an “unauthorized” error. Similarly, a read-token can be used to control accesses that read the tag. Read and write tokens can be independently specified.

When a token is specified for an operation that creates a new tag, that token becomes the new tag’s read and write token. This allows the atomic creation of access-controlled tags.

In addition to being a container of metadata about blobs, a tag can also contain a limited number of user-defined attributes, each with a name and a string value.

Implementation¶

DDFS is embedded in Disco, hence the architecture diagram above closely resembles that of Disco (see Technical Overview). DDFS is currently coordinated by a single master node, similar to Disco itself. This choice was motivated by the ease of implementation and robustness, following experiences of the first version of the Google filesystem. As no data is stored on the master node, it is not a single point of failure with respect to data persistence. It mainly acts as a lock server, ensuring atomicity of metadata operations.

Each storage node contains a number of disks or volumes (vol0..volN), assigned to DDFS by mounting them under DDFS_DATA/vol0 ... DDFS_DATA/volN (see DDFS_DATA). On each volume, DDFS creates two directories, tag and blob, for storing tags and blobs, respectively. DDFS monitors available disk space on each volume on regular intervals for load balancing. New blobs are stored to the least loaded volumes.

Each storage node maintains a cache of all tags stored on the node. When the master node receives a request accessing a yet unseen tag, it queries the storage nodes to find all replicas of the tag. Thanks to the cache, this operation is reasonably fast. Similarly, if the master node crashes and restarts, re-populating the master cache takes only some seconds.

All tag-related operations are handled by the master, to ensure their atomicity and consistency. The client may push new blobs to DDFS by first requesting a set of URLs for the desired number of replicas from the master. After receiving the URLs, the client can push the blobs individually to the designated URLs using HTTP PUT requests. After pushing all replicas successfully to storage nodes, the client can tag the blobs by making a tag request to the master with a list of URLs to the newly created blobs.

If the client fails to push all K replicas to storage nodes, it can request a new set of URLs from the master, excluding the failed nodes. This approach is enabled by default in the DDFS Python API. The client can also decide to accept only M replicas, where M < K, if this is sufficient for the application. If the master detects that a node has become unresponsive, it is automatically blacklisted and dropped from subsequent queries. Thanks to replicated data and metadata, this does not result in any data loss.

A regular garbage collection process makes sure that the required number of replicas is maintained, orphaned blobs are deleted and deleted tags are eventually removed from the system. The desired number of replicas is defined in the configuration file, see disco.settings for details.

Blobs can be accessed either over HTTP, using DDFS’s built-in web server on each storage node, or directly on local disk. The latter feature is heavily utilized by Disco, which prefers to run tasks on the nodes where data is physically stored, to minimize network traffic.

The token-based authorization scheme is implemented using the basic access authentication scheme of HTTP, as described in RFC 2617.

DDFS APIs¶

Internals¶

This section provides information about DDFS internals, supplementing comments in the source code. This discussion is mainly interesting to developers and advanced users of DDFS and Disco.

As one might gather from the sections above, metadata (tag) operations are the central core of DDFS, mainly due to their transactional nature. Another non-trivial part of DDFS is re-replication and garbage collection of tags and blobs. These issues are discussed in more detail below.

Create a DiscoDB¶

First, let’s modify the word count example to write its output to a DiscoDB:

"""
This example could be run and the results printed from the `examples/util` directory in Disco:

disco run wordcount_ddb.WordCount http://discoproject.org/media/text/chekhov.txt
"""
from disco.core import Job
from disco.util import kvgroup

from disco.schemes.scheme_discodb import discodb_stream

class WordCount(Job):
    reduce_output_stream = discodb_stream

    @staticmethod
    def map(line, params):
        for word in line.split():
            yield word, 1

    @staticmethod
    def reduce(iter, params):
        for word, counts in kvgroup(sorted(iter)):
            yield word, str(sum(counts))

Notice how all we needed to do was change the reduce_output_stream to disco.schemes.scheme_discodb.discodb_stream(), and turn the count into a str. Remember, DiscoDBs only store byte sequences as keys and values, its up to the user to serialize objects; in this case we just use str.

Query a DiscoDB¶

Next, lets write a job to query the DiscoDBs.

"""
This example could be run and the results printed from the `examples/util` directory in Disco:

python query_ddb.py <query> <input> ...
"""
import sys
from disco.core import Job, Disco, result_iterator

from disco.worker.classic.func import nop_map
from disco.schemes.scheme_discodb import input_stream

class Query(Job):
    map_input_stream = (input_stream, )
    map = staticmethod(nop_map)

    @staticmethod
    def map_reader(discodb, size, url, params):
        for k, vs in discodb.metaquery(params):
            yield k, list(vs)

if __name__ == '__main__':
    from query_ddb import Query
    job = Query().run(input=sys.argv[2:], params=sys.argv[1])
    for k, vs in result_iterator(job.wait()):
        print('{0}\t{1}'.format(k, sum(int(v) for v in vs)))

Now let’s try creating our word count db from scratch, and querying it:

$ cd disco/examples/util
$ disco run wordcount_ddb.WordCount http://discoproject.org/media/text/chekhov.txt
WordCount@515:66e88:d39
$ disco results @ | xargs python query_ddb.py 'word'
word    18
$ disco results @?WordCount | xargs python query_ddb.py 'this | word'
this | word     217

There are a few things to note in this example. First of all, we use a disco.worker.classic.func.nop_map(), since we do all the real work in our map_reader. We use a builtin disco.worker.task_io.input_stream(), to return a DiscoDB from the file on disk, and that’s the object our map_reader gets as a handle.

Notice also how we turn vs into a list. This is because discodb.DiscoDB.metaquery() returns lazy objects, which cannot be pickled.

Finally, notice how we run the disco.job.Job. We make the input and params runtime parameters, since we are pulling them from the command line. When we iterate over the results, we deserialize our counts using int, and sum them together.

Make sure you understand why we sum the counts together, why there could possibly be more than one count in the result (even though we only had one global reduce in our word count job).

The Job Dict¶

The job dict is a JSON dictionary.

jobdict.pipeline¶

A list of tuples (tuples are lists in JSON), with each tuple specifying a stage in the pipeline in the following form:

stage_name, grouping

Stage names in a pipeline have to be unique. grouping should be one of split, group_label, group_all, group_node and group_node_label.

See also

Pipeline Data Flow in Disco Jobs

jobdict.input¶

A list of inputs, with each input specified in a tuple of the following form:

label, size_hint, url_location_1, url_location_2, ...

The label and size_hint are specified as integers, while each url_location is a string. The size_hint is a hint indicating the size of this input, and is only used to optimize scheduling.

jobdict.worker¶

The path to the worker binary, relative to the job home. The master will execute this binary after it unpacks it from The Job Home.

jobdict.prefix¶

String giving the prefix the master should use for assigning a unique job name.

Note

Only characters in [a-zA-Z0-9_] are allowed in the prefix.

jobdict.owner¶

String name of the owner of the job.

jobdict.save_results¶

Boolean that when set to true tells Disco to save the job results to DDFS. The output of the job is then the DDFS tag name containing the job results.

New in version 0.5.

Job Environment Variables¶

A JSON dictionary of environment variables (with string keys and string values). The master will set these in the environment before running the jobdict.worker.

The Job Home¶

The job home directory serialized into ZIP format. The master will unzip this before running the jobdict.worker. The worker is run with this directory as its working directory.

In addition to the worker executable, the job home can be populated with files that are needed at runtime by the worker. These could either be shared libraries, helper scripts, or parameter data.

The job home is shared by all tasks of the same job on the same node. That is, if the job requires two map task and two reduce task executions on a particular node, then the job home will be unpacked only once on that node, but the worker executable will be executed four times in the job home directory, and it is also possible for some of these executions to be concurrent. Thus, the worker should take care to use unique filenames as needed.

Additional Job Data¶

Arbitrary data included in the job pack, used by the worker. A running worker can access the job pack at the path specified by jobfile in the response to the TASK message.

Creating and submitting a Job Pack¶

The jobpack can be constructed and submitted using the disco job command.

Split¶

The split operation puts each input into its own group, regardless of the label of the input, or the cluster node on which it resides. The grey boxes below indicate the cluster nodes on which the inputs reside, while labels are indicated by the colors of the inputs.

Group_all¶

The group_all operation puts all inputs into a single group, again regardless of their labels or hosting cluster nodes, illustrated below.

Group_label¶

On the other hand, the group_label operation takes into account the labels, and puts all inputs with the same label into the same group. As before, this is done without regard to the cluster nodes hosting the inputs, below.

Map-Reduce¶

To illustrate the pipeline model, we can now use it to express a simple Map-Reduce job, show below, with job inputs and outputs shown in black.

When expressed as a pipeline, this becomes simply:

map-reduce = {split, <map>}, {group_label, <reduce>}

The first stage, <map>, of this pipeline puts each input to the pipeline into its own group, and executes a <map> task with that group as input. Each <map> task produces a set of labeled outputs. These labeled outputs become the labeled inputs of the <reduce> stage. This <reduce> stage puts all inputs with the same label into a single group, and executes a <reduce> task that takes in each such group as input. Hence, the number of such <reduce> tasks is determined by the number of distinct labels generated by the <map> tasks in the <map> stage. The labeled outputs generated by the <reduce> tasks in the <reduce> stage become the outputs of the map-reduce pipeline. This shows how the normal Partitioned Reduce DataFlow in earlier versions of Disco can be implemented.

Let’s now clarify one issue that might be raised at this point: the inputs to the pipeline are not the outputs of a previous stage in the pipeline, so what labels are attached to them? The answer is that the labels for the job inputs are derived from The Job Pack submitted to Disco by the user, and hence are user-specified. (However, see the discussion of backward-compatibility issues in The Job Pack.)

One issue that may be clear from the diagram is that the group_label operation, also known as shuffle in map-reduce terminology, can involve a lot of network traffic. It would be nice to optimize the network transfers involved. Indeed, that is precisely what the remaining grouping operations will let us do.

Group_label_node¶

The group_label_node operation is essentially the group_label operation, but performed on every cluster node.

We can use this operation to condense intermediate data belonging to the same label and residing on the same node, before the data is transmitted across the network.

We can use this in the map-reduce pipeline by adding an additional stage with this grouping operation:

map-condense-reduce = {split, <map>}, {group_node_label, <condense>}, {group_label, <reduce>}

Group_node¶

The last grouping operation, group__node, similarly is essentially the group_all operation, but performed on every cluster node.

This provides an alternative to group_node_label to users when trying to condense intermediate data on a node-local basis. The choice between the two will be determined by the fact that a group_node_label stage will typically have more tasks each of which will process less data (and hence probably have lower memory requirements) than a corresponding group_node stage.

It should be clear now that the five grouping operations provide the user a range of complementary options in constructing a flexible data processing pipeline.

Erlang¶

Unless otherwise specified, Erlang Programming Rules apply.

Also, the use of -spec annotations and Dialyzer is highly recommended before checking in commits.

Python¶

Unless otherwise specified, guidelines from PEP 8 apply.

Documentation¶

• inline with code when possible

Message Format¶

Messages in the protocol, both from and to the worker, are in the format:

<name> ‘SP’ <payload-len> ‘SP’ <payload> ‘\n’

where ‘SP’ denotes a single space character, and <name> is one of:

DONE

ERROR

FAIL

FATAL

INPUT

INPUT_ERR

MSG

OK

OUTPUT

PING

RETRY

TASK

WAIT

WORKER

<payload-len> is the length of the <payload> in bytes, and <payload> is a JSON formatted term.

Note that messages that have no payload (see below) actually contain an empty JSON string <payload> = “” and <payload-len> = 2.

Messages from the Worker to Disco¶

Messages from Disco to the Worker¶

Sessions of the Protocol¶

On startup, the worker should first send the WORKER message, and then request the TASK information. The taskid and mode in the TASK response can be used, along with the current system time, to create a working directory within which to store any scratch data that will not interact with other, possibly concurrent, workers computing other tasks of the same job. These messages can be said to constitute the initial handshake of the protocol.

The crucial messages the worker will then send are the INPUT and OUTPUT messages, and often the INPUT_ERR messages. The processing of the responses to INPUT and INPUT_ERR will be determined by the application. The worker will usually end a successful session with one or more OUTPUT messages followed by the DONE message. Note that it is possible for a successful session to have several INPUT_ERR messages, due to transient network conditions in the cluster as well as machines going down and recovering.

An unsuccessful session is normally ended with an ERROR or FATAL message from the worker. An ERROR message terminates the worker, but does not terminate the job; the task will possibly be retried on another host in the cluster. A FATAL message, however, terminates both the worker, and the entire job.

Considerations when implementing a new Worker¶

You will need some simple and usually readily available tools when writing a new worker that implements the Disco protocol. Parsing and generating messages in the protocol requires a JSON parser and generator. Fetching data from the replica locations specified in the Disco INPUT responses will need an implementation of a simple HTTP client. This HTTP client and JSON tool can also be used to persistently store computed results in DDFS using the REST API. Optionally, there is support to use the Disco master itself to save job-results to DDFS, using the save_results field in the jobdict.

The protocol does not specify the data contents of the Disco job inputs or outputs. This leaves the implementor freedom to choose an appropriate format for marshalling and parsing data output by the worker tasks. This choice will have an impact on how efficiently the computation is performed, and how much disk space the marshalled output uses.