Saturday, April 25, 2009

In Part 1 of this series we introduced a design pattern for grid computing on Azure. In this article we'll implement the pattern by developing a grid application in C# and in Part 3 we'll run the application, first locally and then in the cloud. In order to do that, we'll need some help from a grid computing framework.

The Role of a Grid Framework
Unless you're prepared to write a great deal of infrastructure software, you'll want to use a framework for your grid application that does the heavy lifting for you and lets you focus on writing your application code. While Azure performs many of the services you would want in a grid computing infrastructure, it's still necessary to add some grid-specific functionality between Azure and your grid application. A good grid computing framework should do these things for you:

• Provide a means of scheduling and controlling job runs
• Retrieve input data from on-premise storage
• Generate tasks for grid workers to execute
• Distribute tasks to available workers
• Track the status of tasks as the grid executes the application
• Aggregate results from workers
• Store results in on-premise storage

The diagram below shows how the framework brings the grid application and the Azure platform together. The application developer only has to write application-specific code to load input data, generate tasks, execute tasks, and save result data. The framework provides all of the necessary plumbing and tooling in a way that strongly leverages the Azure platform.



In this article we'll be using Azure Grid, the community edition of the Neudesic Grid Computing Framework. Azure Grid performs all of the functions listed above by providing 4 software components:

• A Loader, to which you add your own code to draw input data from on-premise resources and generate tasks.
• A Worker Role, to which you add your own code to execute application tasks.
• An Aggregator, to which you add your own code to store results back to on-premise resources.
• A Grid Manager, which allows you to start job runs and monitor their execution.

Azure Grid minimizes expense by only using cloud resources during the execution of your grid application. On-premise storage is where input data, results, and Azure Grid's tracking database reside. Cloud storage is used for communication with workers to pass parameters and gather results, and drains to empty as your grid application executes. If you also suspend your grid worker deployment when idle you won't be accruing ongoing charges for storage or compute time once your grid application completes.

The Application: Fraud Check
The application we'll be coding is a fictional fraud check application that uses rules to compute a fraud likelihood score against applicant data. Each applicant record to be processed will become a grid task. The applicant records have this structure:



By applying business rules to an applicant record, the Fraud Check application computes a numeric fraud likelihood score between 0 and 1000, where zero is the worst possible score. An application will be rejected if it scores below 500.

Designing the Grid Application
When you design a grid application you need to determine the best way to divide up the work to be done into individual tasks that can be performed in parallel. You start by considering 2 key questions:

• On what basis will you divide the work into tasks?
• How many different kinds of tasks are there?

In the case of Fraud Check, it makes sense to create a separate task for each applicant record: the fraud scoring for each record is an atomic operation, and it doesn't matter what order the records are processed in as long as they all get processed.

Only one task type is needed for Fraud Check which we'll name "FraudScore". The FraudScore task simply renders a fraud score for an applicant record.
Tasks need to operate on input data and produce results data. The input data for FraudScore will be an applicant record and its results data will be a fraud score plus a text field explaining reasons for the score. FraudScore will expect parameters and return results with the names shown below.



In some Grid computing applications tasks might also need access to additional resources to do their work such as databases or web services. FraudScore does not have this requirement, but if it did some of the input parameters would supply necessary information such as web service addresses and database connection strings.

Developing the Grid Application
Now that our grid application's input parameters, tasks, and result fields are defined we can proceed to write the application. Azure Grid only asks us to write code for the Loader, the application's tasks, and the Aggregator.

Writing the Loader Code
The Loader code is responsible for reading in input data and generating tasks with parameters . Most of the time that will come from a database, but Fraud Check is written to read input data from a spreadsheet.

Azure Grid gives you the following starting point for your Loader in a class named AppLoader. The method GenerateTasks needs to be implemented to pull your input data and generate tasks with your task type names and your parameters. Your code will create Task objects and return them as an array. The base class, GridLoader, takes care of queuing your tasks into cloud storage where they can execute.

#region Application-Specific Loader Code

/// 

/// Your applications' Loader code goes here. 
/// Responsibilities:
/// 1. Read input data necessary to create tasks with parameters from local resources.
/// 2. For each task generated, create a Task object with input parameters.
/// 3. Return an array of Task objects.
/// 

/// Job id
/// Task[] array

public Task[] GenerateTasks(string jobId)
{
    List tasks = new List();
    Dictionary parameters = new Dictionary();

    // TODO: implement Loader

            
    // Example task creation:

    parameters["Param1"] = "Value1";
    parameters["Param2"] = "Value2";
    parameters["Param3"] = "Value3";
    tasks.Add(new Task(ProjectName, jobId, 1, "GridTask1", Task.Status.Pending, parameters, null));

    return tasks.ToArray();
}

#endregion

 

To implement the Loader for Fraud Check, we replace the sample task creation code with this code that reads input records from a spreadsheet CSV file and creates a task for each record.

using (TextReader reader = File.OpenText("FraudInput.csv"))
{
    string[] names = null;
    string[] values = null;

    string line;
    int lineCount = 0;
    int nextTaskId = 1;

    // Read lines from CSV file until empty. Line 1 contains parameter names.

    while ((line = reader.ReadLine()) != null)
    {
        lineCount++;

        if (lineCount == 1)
        {
            // Reader header row of parameter names.
            names = line.Split(',');
            int n = 0;
            foreach (string name in names)
                names[n++] = name;
        }
        else
        {
            if (!String.IsNullOrEmpty(line))
            {
                // Load latest values for this row and generate a task
                values = line.Split(',');

                parameters = new Dictionary();

                for (int i = 0; i < names.Length; i++)
                    parameters[names[i]] = String.Empty;

                for (int i = 0; i < values.Length; i++)
                    parameters[names[i]] = values[i];

                tasks.Add(new Task(ProjectName, jobId, nextTaskId++, "FraudScore", Task.Status.Pending, parameters, null));
            }
        }
    }
}

 

The top row of the input spreadsheet should contain parameter names and subsequent rows should contain values, just as in shown earlier. Creating a task is simply a matter of instantiating a Task object and giving it the following information in the constructor:

• Project Name: Your application's project name. This comes from a configuration file setting.
• Job ID: The Id of this job run, a string. This value is provided to the GenerateTasks method.
• Task ID: A unique identifier for this task, an integer.
• Task Type: The name of the task to run.
• Task Status: Should be set to Task.Status.Pending which indicates a not-yet-run task.
• Parameters: A dictionary of parameter names and values.
• Results: NULL - results will be set by the grid worker that executes the task.

Adding the Task to a List completes the work. Once all of the tasks have been generated, returning the List.ToArray() passes the results to the Loader where they are queued to cloud storage.

Writing the Aggregator Code
The bookend to the Loader is the Aggregator, which processes task results and stores them locally.

Azure Grid gives you the following as a starting point for your aggregator in a class named AppAggregator. There are 3 methods to be implemented:

• OpenStorage is called when the first result is ready to be processed to give you an opportunity to open storage.
• StoreResult is called for each result set that needs to be stored. Both the input parameters and results are passed in as XML.
• CloseStorage is called after the final result has been stored to give you an opportunity to close storage.

The base class, GridAggregator, takes care of processing results from cloud storage and calling your methods to store results.

#region Application-specific Aggregator Code

/// 

/// You application's Aggregator code goes here.
/// Responsibilities:
/// 1. OpenStorage - open local storage.
/// 2. StoreResult - store a result.
/// 2. CloseStorage - close local storage.
/// 

/// 

protected override void OpenStorage()
{
    // TODO: open storage
}

protected override void StoreResult(string parametersXml, string resultsXml)
{
    // TODO: store result
}

protected override void CloseStorage()
{
    // TODO: close storage
}

#endregion

 

In StoreResult, both the parameters and results for the current task are passed in as XML in this format:

<Parameters>=>
  <Parameter name="LastName" value="Bach"/>
  <Parameter name="FirstName" value="J.S."/>
  ...
</Parameters>
<Results>
  <Result name="Score" value="700"/>
  <Result name="Approved" value="1"/>
  <Result name="Notes" value=" "/>
</Result>

 

To implement the aggregator for Fraud Check, we'll reverse what the Loader did and append each result to a spreadsheet CSV file.

• In OpenStorage, a .csv file is opened for output and the column row of the spreadsheet CSV file is written out.
• In StoreResult, results (and also the first and last name input parameters to provide context) are extracted from XML and written out.
• In CloseStorage, the file is closed.

#region Application-specific AggregatorCode

/// 

/// You application's Aggregator code goes here.
/// Responsibilities:
/// 1. OpenStorage - open local storage.
/// 2. StoreResult - store a result.
/// 2. CloseStorage - close local storage.
/// 

/// 

TextWriter tw = null;

protected override void OpenStorage()
{
    tw = File.CreateText("FraudOutput.csv");
    tw.WriteLine("Last Name,First Name,Score,Accepted,Notes");
    tw.Flush();
}

protected override void StoreResult(string parametersXml, string resultsXml)
{
    XElement parameters = XElement.Parse(parametersXml);
    XElement results = XElement.Parse(resultsXml);

    // Write values

    tw.WriteLine("{0},{1},{2},{3},\"{4}\"",
        parameters.XPathSelectElement("/Parameter[@name='LastName']").Attribute("value").Value,
        parameters.XPathSelectElement("/Parameter[@name='FirstName']").Attribute("value").Value,
        results.XPathSelectElement("/Result[@name='Score']").Attribute("value").Value,
        results.XPathSelectElement("/Result[@name='Accepted']").Attribute("value").Value,
        results.XPathSelectElement("/Result[@name='Notes']").Attribute("value").Value);

    tw.Flush();
}

protected override void CloseStorage()
{
    if (tw != null)
        tw.Close();
}

#endregion

 

Writing the Application Task Code
With the loader and aggregator written, there's just one more piece to write: the application code itself. The AppWorker class contains the application task code. The current task is passed to a method named Execute is which examines the task code to determine which task code to execute.

#region Application Code

/// 

/// Application code to execute a task. The switch statement uses the task's TaskType string to determine the appropriate code to execute.
/// 

/// The task to execute.

public override void Execute(Task task)
{
    switch (task.TaskType)
    {
        case "Task1":
            Task1(task);
            break;
        case "Task2":
            Task2(task);
            break;
        default:
            // Ignore unknown task.
            break;
    }
}

private void Task1(Task task)
{
    // TODO: implement task1
}

private void Task2(Task task)
{
    // TODO: implement task 2
}

#endregion End Application Code

 

For Fraud Check, the switch statement checks for the one task type in our application, FraudScore, and executes the code to compute a fraud likelihood score based on the applicant data in the input parameters.

public override void Execute(Task task)
{
    switch (task.TaskType)

        case "FraudScore":
            FraudScore(task);
            break;

        default:
            // Ignore unknown task.
            break;
    }
}

 

The first order of business for the FraudScore code is to extract the input parameters, which are accessible through a dictionary of names and string values in the Task object.

private void FraudScore(Task task)
{
    StringBuilder notes = new StringBuilder();

    bool rejected = false;
    int score = 1000;

    string firstName = task.Parameters["FirstName"];
    string lastName = task.Parameters["LastName"];
    string state = task.Parameters["State"];
    string country = task.Parameters["Country"];
    int age = Convert.ToInt32(task.Parameters["Age"]);
    string ssn = task.Parameters["SSN"];
    string relation = task.Parameters["Relation"];
    int monthsEmployed = Convert.ToInt32(task.Parameters["MonthsEmployed"]);

 

Next, a series of business rules execute that compute the score. Here's an excerpt:

    // Rule: if age < 18 or age > 100, automatic rejection

    if (age < 18 || age > 100)
    {
        score = 0;
        notes.Append("Age out of range, automatic rejection. ");
        rejected = true;
    }

    // Rule: is SSN missing, reduce score by 300.

    if (string.IsNullOrEmpty(ssn))
    {
        score -= 300;
        notes.Append("SSN missing. ");
    }

...

    // Check score. If below 500, reject application.

    if (score < 0) score = 0;

    if (!rejected && score < 500)
    {
        rejected = true;
        notes.Append("Score below 500, rejection. ");
    }

 

Lastly, FraudScore updates the task with results. This is simply a matter of setting names and string values in a dictionary.

    // Store task results.

    task.Results["Score"] = score.ToString();
    task.Results["Notes"] = notes.ToString();

    if (rejected)
        task.Results["Accepted"] = "0";
    else
        task.Results["Accepted"] = "1";
}

 

The base GridWorker class and WorkerRole implementation take care of queuing the results to cloud storage where they will be retrieved by the Aggregator.

Ready to Run
We've developed our grid application and are about ready to run it. Just a quick review of what we've just accomplished: using a framework, we implemented a loader, an aggregator, and task code. We only had to write code specific to our application.

All that remains is to run the application. With a grid application, you should always test carefully, initially by running locally with a small number of tasks. Once you're confident in your application design and code integrity, you can move on to large scale execution in the cloud. We'll be doing just that in the next article in this series, Part 3.