Docs: Computation Layer

The Computation Layer performs some of the functions that the Serving Layer can perform when run it its stand-alone mode:

• Storing received input data from the Serving Layer
• Computing and recomputing the underlying recommender model
• Publishing the latest model for use by the Serving Layer

and some new functions:

• Discovering clusters of users and items

However, in the Computation Layer, the storage and computation are distributed, using computers to process much more data, faster. It is implemented on top a distributed storage system, and uses Apache Hadoop, an implementation of the MapReduce distributed computing paradigm. This document covers the Computation Layer itself, and how it may be run independently of the Serving Layer if desired.

Cluster Requirements

The Computation Layer requires a Hadoop cluster and a distributed file system. Two popular deployments of this infrastructure are supported. First is an existing Hadoop cluster, with its associated HDFS file system, which may already be available at your organization. Second is Amazon's Elastic MapReduce (EMR) Hadoop service and Simple Storage Service (S3) distributed storage system. In this model, Amazon creates a Hadoop cluster on demand, which is billed by the hour.

Supported Hadoop cluster versions

Supported Amazon AWS versions

Cluster Configuration

See the Tuning Guide for suggested cluster configuration settings. This does not apply to Amazon EMR, where Myrrix can configure the cluster for you directly.

Storage Layout

The Computation Layer reads input from the distributed storage system, uses it for intermediate results, and outputs the result of each run to the same distributed storage system. This section describes how the files and data are laid out on the storage system.

Instances

All files are under one root directory, called the bucket, which should be unique to the Myrrix user calling the Computation Layer. For example, Acme Corporation might use "acme". This corresponds to a root-level directory in HDFS (e.g. /acme), or a storage bucket in Amazon S3.

Each bucket is further divided to accommodate independent instances of the Computation Layer. For example, maybe Acme is running three recommenders using Myrrix, for different departments. These different instances are identified by a unique ID. A separate directory under the root directory is created and used for each instance. For example, Acme's instance ads uses files only under /acme/ads. In Amazon's S3, this would correspond to key "ads" under bucket "acme". For simplicity, going forward, this document will refer to HDFS-style file-system-like paths instead of using S3's bucket and key naming.

Generations

Computation proceeds in generations. Input is collected from Serving Layers, and periodically, a new computation generation is run using all input received to that point. Each generation uses a new sequential sub-directory of the instance directory. For example, the first time Acme instance ads runs, it will create generation 0 in /acme/ads/0000000000. Subsequent generations 1, 2 and so on will follow in sequence, in .../0000000001 and so on.

Under a generation's directory, the Serving Layers (or any other process) write input into files in an inbound subdirectory (example: /acme/ads/0000000001/inbound). All files in this directory should be CSV-format text files, optionally compressed. Processes may copy files into this directory. Files should be closed and not be modified when a generation's computation starts.

The Computation Layer runs a generation. It is responsible for creating new future generations, which other processes should detect and start writing into, so that the Computation Layer may proceed on the current generation. When it finishes, the Computation Layer publishes its output -- the factored X and Y matrices, in X/ and Y/ subdirectories (example: /acme/ads/0000000001/X). These are (compressed) text files. Each line of each file contains a row or column index (that is, a user ID for X, or item ID for Y), followed by a tab, then a series of comma-separated floating-point values, which are the feature values for that row/column in X/Y. Example:

123 0.352008,-0.11290,0.982342,...

Optionally, the Computation Layer can also create recommendations for all users. This is normally produced by the Serving Layer in real-time, but can also be computed by the Computation Layer. This may be useful when only using the Computation Layer, by itself. The --recommend flag will cause it to output recommendations in a recommend subdirectory (e.g. /acme/ads/0000000001/recommend). This also contained compressed text files. Each line has a user ID, followed by tab, the comma-separated itemID:score pairs:

456 5802:0.9855,2059:0.97001,...

The Computation layer can also compute the most similar items for all items. This is controlled by the --itemSimilarity flag, and produces output in similarItems/. The format is identical to the recommendations above.

Finally, the Computation Layer can compute user and item clusters using the k-means++ algorithm when --cluster is set. These are written as compressed text files to userClusters and itemClusters subdirectories. The format is the user or item ID, followed by tab, followed by a cluster number.

Adding input

Input is normally added through a Serving Layer instance that is connected to the Computation Layer. Data that arrives to the pref or ingest method is copied into the latest generation's inbound/ directory. It is also possible to copy CSV input directly into the same location on HDFS, into the most recent generation's inbound/ directory. The format is same as is used in Apache Mahout and the same that is used with the ingest method.

Running the Computation Layer

The Computation Layer is run as a Java program. It is simply a client that invokes the computations on a Hadoop cluster (or Amazon EMR), monitors them, and performs other operations like cleanup. It exposes a web-based UI which gives status and information on its progress. It runs continuously and periodically creates and runs new computation generations. It can be configured to run new generations after an amount of time has passed, or after an amount of data has been written, or both.

A simple invocation looks like:

java -jar myrrix-computation-x.y.jar --bucket acme --instanceID ads --licenseFile ...

This runs a Computation Layer that manages bucket "acme", instance "ads" (that is, data in /acme/ads). A web-based UI is run by default on port 8080, and can be accessed on the local machine at http://localhost:8080.

The following command line arguments are available:

The following system properties may also be set on the command-line using Java's -Dkey=value syntax. These must appear before -jar. These are advanced settings that are not necessary to change.

Example:

java -Doutput.recs.count=100 -jar myrrix-computation-x.y.jar --bucket acme --instanceID ads --time 12h --recommend --port 80 --licenseFile ...

This runs a Computation Layer for bucket "acme" and instance ID ads, as before. It computes 100 recommendations for all users with each generation. Every 12 hours it will attempt to run a new generation. The web based UI runs on port 80 (note that it must run as root if using port 80).

Platform-specific Notes

Notes on Apache Hadoop

The following system properties may be set to affect how the Computation Layer runs:

Notes on Amazon EMR

When using Amazon EMR, use the JAR file built to work with Amazon's Web Services APIs, myrrix-computation-aws-x.y.jar. Note that the client will upload the JAR file itself to S3 at sys/myrrix.jar since it must be available remotely.

It is necessary to supply an AWS access key and secret key that is authorized to use the bucket, and authorized to run jobs on AWS. These are specified as system properties. For example:

java -Dstore.aws.accessKey=[access key] -Dstore.aws.secretKey=[secret key] -jar myrrix-computation-aws-x.y.jar --bucket acme --instanceID ads

They may also be specified as the standard AWS environment variables, AWS_ACCESS_KEY and AWS_SECRET_KEY.

The complete set of system properties available that control the behavior of the Computation Layer on Amazon S3 and EMR are: