I'm on a mission from God object : HPC

LINQ to HPC (Formerly known as DryadLINQ) Tutorial: Part 2–Data Partitioning (DSC)

Yaniv Rodenski — Tue, 09 Aug 2011 18:39:00 GMT

A new beta has been released since I wrote part 1 of this tutorial. While very little was changed in the product, we have a new name. Another thing held me back personally from publishing this part was the fact that LINQ to HPC is not a part of Windows HPC R2 SP2. So without farther ado I am proud to present the second part of my tutorial about LINQ to HPC.

In part 1 of this tutorial we discussed the fundamentals of DSC: how to manually write data to DSC files and how to use the FromEnumerable<T> extension method (from the HpcLinqExtras project) to implicitly save object data to a temporary file set (in order to use it inline in a subsequent query). We also saw a caveat in this method, namely that because FromEnumerable<T> saves the data to a single file in the temporary file set, the subsequent query cannot be parallelized. This is due to the fact that LINQ to HPC runs any query logic locally on the DSC node containing the data to which it refers.

The task at hand is quite straight forward: we would like to partitions our data into logical pieces that can be distributed across the cluster. Before we start discussing how we can physically partition data in LINQ to HPC, I would like to consider the logic we will use for dividing the data into groups. in order to do so we will take a look at vertices, which are the basic tasks that execute the query on the cluster. I will describe vertices in detail in a later part of this tutorial but for now there are few facts I would like you to consider:

A vertex can only use data from a single DSC file, located on the node it is executing on. This is, of course, in order to preserve data locality. The main implication of this little fun fact is that we should make sure that pieces of data that are dependent on each other will reside continuously in the DSC file set. A good example for this is the use of GroupBy in a query. Lets create a Student class defined as follows:

[Serializable]
public class Student
{
    public int Id { get; set; }
    public string Name { get; set; }
    public string Nationality { get; set; }
    public double AvgGrade { get; set; }
}

Now let’s say we are grouping our Persons by nationality, so our data should be ordered like this:

Dryad can execute local queries in each vertex and then union all the groups. If the same data needs to be reordered by the query (let’s say items were ordered by Id in the query), the first thing LINQ to HPC would need to do is to reorganize the data into intermediate files, and only then execute the necessary logic.
Note: grouping operators are a bit more complex when it comes to LINQ to HPC and will be discussed in a later part of this tutorial.
A vertex will process all the data in the DSC file it is accessing. This means that if we would like to break down the processing of local queries in to smaller pieces we need to break the data in to smaller files. This is possible since DSC file set support creating more files than the number of nodes.

We can control the order in which our objects are written to file when using custom HPC serialization (as I have shown in part 1 of the tutorial). However this can become tedious, especially if we need to use the same data in different queries that can benefit from different partitioning and ordering.

Repartitioning Operators

Repartitioning operators are LINQ to HPC operators that result in intermediate DSC files partitioned in a way that is not dependent on the partitioning of the input files. There are two Repartitioning operators in LINQ to HPC: Hash and Range Partitioning.

Hash Partitioning

Hash partitioning provides a mechanism for partitioning data that is not sorted; Returning to our students sample, nationality is a prime candidate for hash partitioning. To use hash partitioning you need to call the HashPartition operation, which provides an overload that accepts the number of partitions to be created, once called you can use the ToDsc operator to create a new DSC file set and call SubmitAndWait to commit the operation (I have reviewed this steps in part 1 of this tutorial):

// getting the list of students
List<Student> students = GetStudentsList();

// saving the students range partitioned to the file set with 5 partitions
context.FromEnumerable<Student>(students)
       .HashPartition(std => std.AvgGrade, 5)
       .ToDsc<Student>("StudentsFileSet")
       .SubmitAndWait(context);

The Why hash partitioning selects the partition for a specific entity is by performing a mod operation between the hash code of the key selector and the number of partitions, the following code mimics the behavior of hash partitioning regarding the partition selection:

This method is disappointingly crude. If you run this code (supplied with my samples) you will see that although we have instructed the HashPartition operator to create 5 partitions, the result of the mod operation results in only four different values. This is of course due to the nature of the values in our key selector (none of them divides evenly by 5). This result is somewhat arbitrary, and we could have had the result distributed in many ways (even and un even) dependent on the result of the key selector GetHashCode. To overcome this pitfall, HashPartition has another overload that accepts an IEqualityComparer that can be used to override the implementation of GetHashCode of the key selector.

Range Partitioning

Range partitioning allows the ordered partitioning of sorted keys. Returning once more to our student’s sample, the average grade can be used as such a key. This is useful if our query uses this key selector ordering in its logic. The way range partitioning works is by assigning a range of keys for every file: any object whose key belongs in that range will be placed in the DSC file. By using this method files can be created un-evenly, but we can ensure that objects within a specific range will reside in the same file. Range separators are used to define ranges: these are values that mark the border points between one range and another. Let’s say we now would like to partition our students into files that are partitioned by grades. We will use two range separators to split the data in to three files:

In this case our range separators are 3 and 6. One thing that is very easy to overlook is the fact that if our student’s grade equals the value of a range separator, it can belong, range-wise to the two files on both sides of the separator. Range separators can be assigned in two ways:

Statically assigned by user:
In some cases we would like to explicitly force the range structure. This is useful when we know our data and queries structure and believe we can benefit from it. Let’s say we know our queries mostly filter students with grades of 6 and above, we can reflect this knowledge into our file structure even dough it results in an uneven distribution.
We can pass an array of range separators like this:

// getting the list of students
List<Student> students = GetStudentsList();

// saving the students range partitioned to the file set
context.FromEnumerable<Student>(students)
       .RangePartition(std => std.AvgGrade, new[] { 3d, 6d })
       .ToDsc<Student>("StudentsFileSet")
       .SubmitAndWait(context);

All we need to provide here is a key selector delegate, to select the value on which we partition and the rangeKeys parameter which holds the array of range separators of the same type as the return type of the key selector.
Dynamically sampled:
Another, perhaps simpler approach is to use a different overload that allows LINQ to HPC to generate partition separators for us. When we allow RangePartition to select the range separators for us, it will try to create DSC files of approximately equal size, but on the other hand we do lose much of the control we had creating the range separators ourselves. There are few overloads of RanePartition; the simplest looks like this:

// getting the list of students
List<Student> students = GetStudentsList();

// saving the students range partitioned to the file set with 5 partitions
context.FromEnumerable<Student>(students)
       .RangePartition(std => std.AvgGrade, 5)
       .ToDsc<Student>("StudentsFileSet")
       .SubmitAndWait(context);

Other than losing control with dynamic range partitioning there are few key points you should bear in mind:
- Currently dynamic sampling will take place for every 1,000 records - not really useful for small datasets.
- Dynamic range partitioning is using range separators even if you did not set them yourself. If the key selector will return non-proportional ranges, the files will have to differ in size.

Summary

Data partitioning allows us to implicitly distribute our data over the cluster, thus adding more control to how (and where) our queries will execute. Now that we got all our data just where we want it, we can start creating distributed kick-ass queries. But this calls for a completely different post.

Source code for all the samples can be found here.

DryadLINQ Tutorial: Part 1 – Distributed Storage Catalog (DSC) Basics

Yaniv Rodenski — Tue, 14 Jun 2011 22:29:59 GMT

One of the most exciting additions to Windows HPC Server 2008 R2 SP2 (currently in beta) is the support for DryadLINQ. DryadLINQ is an API that allows the creation and execution of large scale, data-parallel compute tasks. One of the core capabilities of Dryad (the underlying framework used by DryaLINQ) is the ability to distribute the data over the cluster and maintain data locality by executing the code on the node storing the data. In order to do both, Dryad utilizes a mechanism called The Distributed Storage Catalog (DSC) which I will cover in this post.

Overview

DryadLINQ provides a powerful mechanism for distributing LINQ queries over a cluster. In order to do so, Dryad needs to distribute both the code to be executed and the data on which it needs to operate. Before we can execute DryadLINQ queries over any type of data we must save it to DSC. This is what allows DryadLINQ to execute parts of the query (called vertices) in a distributed manner: each vertex will then emit a result that is either retuned to the caller or saved to DSC for further querying.

The HPC Dsc Service

When you install a head node using SP2, a new Windows service - HPC Dsc – is added (Note: for the current beta version of SP2, DryadLINQ requires a new installation, no upgrades are supported). HPC Dsc exposes a service API (via net.tcp binding) with two primary functions:

Provide logical management for DSC entities such as as DSC nodes, file sets, files, etc. (we will discuss those shortly) in a database called HPCDsc.
Physically manage the participating nodes’ files by accessing the file system via a file share.

Figure 1: Overview of the HPC Dsc service

In order to allow DSC on any node simply run the following command:

dsc node add [compute node name] /tempath:[local path for HpcTemp share] /datapath:[local path for HpcData share] /service:[head node name]

This will create the shared directories on the on the target node, and register them with the HPC Dsc service. Once this command is successfully executed, the corresponding compute node can be used for DSC (thereby enabling Dryad) operations.

Working Directly With DSC Files

DSC allows the storage of either text files or serialized objects. The most straightforward way to save either is to write directly to DSC files as the following code demonstrates (you can download all code samples for this post from here):

// creating a fileset and two files manually
if (context.DscService.FileSetExists("TextFileSet"))
    context.DscService.DeleteFileSet("TextFileSet");

DscFileSet fileSet = context.DscService.CreateFileSet("TextFileSet");

DscFile fileA = fileSet.AddNewFile(40000);
DscFile fileB = fileSet.AddNewFile(40000);

// copying the content of text files in to the DSC files
File.Copy(Path.Combine(Directory.GetCurrentDirectory(), @"TextFiles\TextFile1.txt"),
          fileA.WritePath);
File.Copy(Path.Combine(Directory.GetCurrentDirectory(), @"TextFiles\TextFile2.txt"),
          fileB.WritePath);

fileSet.Seal();

var lines = context.FromDsc<LineRecord>("TextFileSet");
Console.WriteLine("The number of lines is {0}", lines.Count());

in this snippet I used a couple of the DSC entities (or better said, their .NET APIs which reside in the Microsoft.Hpc.Dsc namespace) and I would like to review them before we continue:

DscFileSet:
The DscFileSet provides a mechanism to access, manage and most importantly query a group of files distributed over the cluster.
DscFile:
These represent actual files in the FileSet and provide a few useful properties including the WritePath I used in the code above to simply write to the files.

In the above snippet we use the DscFileSet to create two DscFiles. The DscFileSet creates each file in the DscData shared directory on whichever DSC node it sees fit to host the file. We can influence this decision by passing a preferred node name to the AddNewFile method. Once created, we can then manually write to the DscFile.WritePath. When we are done writing to the files we call the Seal method. The Seal method informs the HPC Dsc service that we are done writing to the files. This means that the HPC Dsc service can now update the databse with the file sizes of each file, and change the state of the DscFileSet to sealed.

Another API I used in this snippet is the (declared outside of the presented code) HpcLinqContext (from the Microsoft.Hpc.Linq namespace) which I will cover in more details in later parts of this tutorial. For now, however, I’ll just explain its role in the above code: Once configured the HpcLinqContext allows us to communicate with the DSC in couple of ways. The first one is by exposing the HPC Dsc service via the DscService property: we use the Dsc service to create the new file set. The second way is by allowing us to execute a DryadLINQ query against the FileSet using the FromDsc<T> method.

Saving Serialized Objects To Files

As I mentioned above, DSC supports serialized objects. While this ability exposes the real strength of Dryad, it is unfortunately a mess in its current state (SP2 beta). Fortunately, the DryadLINQ samples that are a part of the SP2 installation contain a project named HpcLinqExtras which simplify much of the oddities of DSC in its current state. I will review these oddities and their solutions in the HpcLinqExtras.

So lets begin by defining a class named Person and mark it as [Serializable]. There are some restrictions for objects that can be serialized to DSC that can be found in the DryadLINQ and DSC Programmer’s Guide, but for our sample we will use the following class definition, and you’ll just have to take my word that it is a complaint with regards to the DSC restrictions:

[Serializable]
public class Person
{
public int Id { get; set; }
public string Name { get; set; }
}

One would believe this should be enough for saving instances of Person to DSC, but with DSC nothing is as simple as it seems. The biggest DSC serialization crankiness is the need for a type implementing custom HPC serialization. Note that not all types in the graph saved need to implement custom HPC serialization - only the root type that will be used for saving manually the data. The HpcLinqExtras project contains the ObjectRecord class that can be used as a sort of generic root type for any .NET object. To implement custom HPC serialization, a type must be decorated with the CustomHpcSerializerAttribute and implement the IHpcSerializer<T> interface, this will allow it to read and write the type passes as the T type parameter from and to DSC. The ObjectRecord for example class is defined as follows:

[CustomHpcSerializer(typeof(ObjectRecord))]
public class ObjectRecord : IHpcSerializer<ObjectRecord>

The next step is to implement a serialization mechanism for writing the actual data to by writing the two method declared in the IHpcSerializer interface: Read and Write. This can be done by simply using runtime serialization into and out of memory streams and passing the latest to the HpcBinaryWriter and HpcBinaryReader. Since the same behavior needs to be invoked manually (without using the HPC binary reader/writer) the object record exposes overloads that use a BinaryWriter and a BinaryReader for the Read and Write methods.

Now we can create new instances of Person and write our own objects one by one, wrapped in ObjectRecords:

// creating a fileset and two files manually
if (context.DscService.FileSetExists("PersonFileSet"))
    context.DscService.DeleteFileSet("PersonFileSet");

DscFileSet fileSet = context.DscService.CreateFileSet("PersonFileSet");

DscFile fileA = fileSet.AddNewFile(100);
DscFile fileB = fileSet.AddNewFile(100);

var andy = new Person { Id = 1, Name = "Andy" };
var kelly = new Person { Id = 2, Name = "Kelly" };

// writing each person to a different file
using (BinaryWriter bw = new BinaryWriter(File.OpenWrite(fileA.WritePath)))
{
    new ObjectRecord(andy).Write(bw);
    bw.Close();
}

using (BinaryWriter bw = new BinaryWriter(File.OpenWrite(fileB.WritePath)))
{
    new ObjectRecord(kelly).Write(bw);
    bw.Close();
}

fileSet.Seal();

Once saved we can query the DscFileSet using the FromDsc<ObjectRecord> to get them back, and use a simple Select to return IQueryable<Person>:

var champs = context.FromDsc<ObjectRecord>("PersonFileSet")
.Select<ObjectRecord, Person>(or => or.Value as Person);

Saving Serialized Objects To Files – Take 2

The above code might seem reasonable when working with two instances and a small FileSet. But what if we have a large collection, and we want to distribute it over the cluster? Well our good friend HpcLinqExtras provides a couple of nifty extension methods that make all this pain go away. For starters we can use the FromEnumerable<T> to write our data structure to a temporary DSC file set and return the good old IQueryable<T> that we can save using the ToDsc<T>:

var andy = new Person { Id = 1, Name = "Andy" };
var kelly = new Person { Id = 2, Name = "Kelly" };
var source = new List<Person>();
source.Add(andy);
source.Add(kelly);

context.FromEnumerable<Person>(source)
.ToDsc<Person>("PersonFileSet")
.SubmitAndWait(context);

So, let’s just see what we have, shall we?

First we call the FromEnumerable<Person> method that works in the following manner:
- It creates a temporary fileset with one file
- Then, the FromEnumerable method loops over the IEnumerable it receives as a parameter and wraps each member in an ObjectRecord for writing to the file.
- Finally the FromEnumerable<Person> calls FromDsc<ObjectRecord>, and uses a simple Select to return IQueryable<Person>.
Next, we use the ToDsc<Person> call to create a file set named “PersonFileSet”, the result of the IQueryable<Person> we just got as a result from the FromEnumerable<Person> method. You might have noticed that ToDsc<T> has that magical ability to write runtime serializable and not just custom HPC serializable objects, I’ve warned you, the API for working with serialized objects is a mess.
Finally, we call the SubmitAndWait method which obviously submits a job to the head node, and then, wait for it… it waits for it to complete.

You might get the feeling we are no better off then before. Granted, we wrote a lot less code then in the previous version but not only that FromEnumerable<T> does exactly what we’ve done in our first sample, we then query the temporary file set and write it again to the final file set. Astute readers might even notice that both times the file sets contain a single file: For the temporary file set a single file was created manually, and if you take a closer look at the sample you will notice that the “PersonFileSet” was created implicitly using the ToDsc<Person> method. So we are not even distributing our data. There is a way to distribute data using this technique, called partitioning which I’ll cover in the next part of this tutorial.

Summary

The fact that DSC, and the Dryad stack are still in early stages doesn't undermine the fact that we are looking at a technology that can change the face of parallel application. And while it is likely some (if not all) of the extensions available today in the HpcLinqExtras will find their way in to the final release of DryadLINQ, it is important to understand how it works and what are the implications that come with it.

There's nothing like a job well done

Yaniv Rodenski — Thu, 14 Apr 2011 13:21:00 GMT

Today @ MIX Ido and I got some very good news. The first part of our work for the HPC Azure Burst training kit was released today for download by Microsoft. Oh and we got a Kinect.

The document released provides on overview of the architectural considerations for using Azure compute nodes as a part of your cluster.

There is a lot of work still ahead of us and we will kick out some very cool content on MSDN (and cooler stuff on our blogs ) so stay tuned.

Yaniv

Understanding HPC SOA application

Yaniv Rodenski — Sun, 10 Apr 2011 21:00:00 GMT

As server developers, we are used to a certain level of interactivity. Our services get a request and often return some kind of response. Lately I found myself justifying having HPC batch jobs. It’s sometimes hard to grasp that classic HPC programs, like the Human Genome Project, rendering a full feature 3D animation movie, or simply operating a “civilian purposes” nuclear reactor, take a long time. And when you execute such programs for days, weeks, months, etc., interactivity is not even a consideration.

Batch jobs are the very essence of classic HPC applications. Still, Microsoft is trying to expose Windows HPC Server 2008 to new verticals, some of which would like to use it for more interactive tasks. For such purposes Windows HPC Server 2008 supports an SOA model exposed via WCF. Microsoft also released a cluster debugger that has some cool features such as:

Cluster debugging
Local Debugging
Running service code locally in a simulated Windows Azure environment
Two project templates for creating both Interactive and Durable Session clients

There is also a decent MSDN walkthrough that shows how to create and debug an HPC SOA application. I recommend running through it, to get a feeling for how HPC SOA applications are built. In this post, I would like to take a deeper look at both session types and some of the mechanisms and techniques they utilize, both on the client and in the cluster.

Sessions

Sessions are an essential part of HPC SOA clients. There are two types of sessions: Interactive and Durable sessions. Neither HPC client session type provides the same semantics that WCF sessions provide, where every call during the session uses the same instance of the service class on the server side. In fact, one of the roles of sessions, when used correctly, is to ensure that calls during the session will be load-balanced between different compute nodes in the cluster. Conversely, durable sessions provide another key functionality which I will describe below. To understand this process we need to take a look at two more components: The Job Scheduler and Broker Node.

The Job Scheduler

The job scheduler is the main component that runs on the head node. It handles units of work called jobs. The main concern of the job scheduler is to allocate the necessary resources for the job and start sub-units of the job tasks on the allocated compute nodes. In an SOA application, the tasks that the job scheduler creates are called service tasks and they host the services defined in the job.

Figure 1: A session starting service tasks using the job scheduler

In SOA applications the job scheduler has one more task, start a broker node that will be used to load balance all service calls between the service tasks.

The Broker Node

The broker node provides a few key capabilities for SOA applications, the first is exposing an endpoint for every service call targeting the specific service job. Every call that is sent to the broker node through that endpoint, will be load balanced between the service tasks in that job.

Figure 2: A Windows HPC Service Message Lifecycle

And now that we understand the basic stuff we can look into two of the more powerful (and cool) scenarios HPC provides for SOA applications.

Durable Sessions

By now we discussed mechanisms that work the same way in both Interactive an Durable sessions. The main difference between both sessions is (not surprisingly) durability. Durable sessions simply save the response message in an MSMQ queue on the broker node where they can be retrieved by clients, either the initiating client or any other client with the permissions to attach itself to the session. Another difference is that in order to use durable sessions, one must use the BrokerClient class in order to send and receive messages defined as MessageContracts as shown in the following snippet:

using (DurableSession session = 
        DurableSession.CreateSession(info))
 {
     Binding binding = new BasicHttpBinding();
 
     using (BrokerClient<ISquareService> client =
         new BrokerClient<ISquareService>(session, binding))
     {
         // Set the response handler
         client.SetResponseHandler<SquareResponse>((response) =>
         {
 
             int reply = response.Result.SquareResult;
             Console.WriteLine("Received response for request {0}: {1}",
                   response.GetUserData<int>(), reply);
         });
 
         client.SendRequest<SquareRequest>(
             new SquareRequest(1000 + i), i);
     }
 }

Grow/Shrink

Grow/Shrink is basically a scheduling policy, and while the job scheduler provides few of these, Grow/Shrink is the more relevant for SOA applications. As its name suggests, Grow/Shrink allows administrators to add or remove resources for a job over time. This helps dealing with peeks and can even be done using Windows Azure Worker Roles as additional compute nodes. The cool thing is that once you add the resources to the job, the job scheduler notifies the broker node about the new resources and the broker node can now take them into account while load balancing.

Conclusion

SOA is still the the ugly duckling in the world of HPC. In fact the whole concept of using sessions to start a service on the cluster comes from the classic HPC paradigm of a single job that needs to be distributed on a cluster. In fact, you can start a service job directly on the server using The HPC 2008 Cluster Manager or use some very simple patterns to start a session that will be shared among all clients transforming Windows HPC Server 2008 R2 in to a lean, mean SOA machine.

There are many capabilities just waiting to be used in commercial SOA applications and the hardware does not need to be a monstrosity (but that can definitely make things cooler). The constant improvement in Windows Azure support makes this type of solutions ideal for dealing with uncommon / unpredicted peeks, since there is no need to buy the hardware to support them.

So until next time, remember: real developers use big computers.

Yaniv