Carl's Blog

Hadoop .Net HDFS File Access

2013-02-08T18:22:45Z

Provided with the Microsoft Distribution of Hadoop, HDInsight, is a C library for HDFS file access. This code extends this library through a Managed C++ solution. This solution enables one to consume HDFS files from within a .Net environment. The purpose of this post is first to ensure folks know about the new Windows HDFS Managed library (WinHdfsManaged) provided alongside the native C library, and secondly to give a few samples of its usage from C#.

The complete code libraries can be downloaded from here: http://code.msdn.microsoft.com/Hadoop-Net-HDFS-File-Access-18e9bbee

Class Structure

Let’s start with a simple class diagram of the Win HDFS Managed library:

The main premise is that the HdfsFileSystem is your starting point, from which one can acquire a HdfsFileStream or a HdfsFileHandle. From the HdfsFileStream you can perform operations one would normally expect when working with .Net Streams. From the HdfsFileHandle you can perform operations analogous to normal HDFS file operations.

For brevity I have excluded samples using the HdfsFileHandle. So let’s run through some sample file operations.

Directory Operations

As in all operations one firstly needs to get a connection to the HDFS cluster. This is achieved by calling a Connect() method and specifying the host, name or IP address, and access port:

Create File System Access

using (HdfsFileSystem hdfsSystem = HdfsFileSystem.Connect("127.0.0.1", 9000))
{
...
}

Once one has the connection one can then easily perform a directory traversal to enquire into the files and directories:

List Directory Structure

Action<string> processDirectory = null;
processDirectory = (looppath) =>
{
using (HdfsFileInfoEntries entries = hdfsSystem.ListDirectory(looppath))
{
foreach (HdfsFileInfoEntry entry in entries.Entries)
{
string kind = entry.Kind == HdfsFileInfoEntryKind.Directory ? "Directory" : "\tFile";
Console.WriteLine(string.Format(@"{0}:""{1}"", Modified/Accessed:""{2:G}, {3:G}"", Owner:""{4}""", kind, entry.Name, entry.LastModified, entry.LastAccessed, entry.Owner));
if (entry.Kind == HdfsFileInfoEntryKind.Directory)
{
processDirectory(entry.Name);
}
}
}
};
processDirectory(hdfspath)

Here is a sample output created from the test application:

Directory:"hdfs://127.0.0.1:9000/user/isotope/qwanchi", Modified/Accessed:"30/01/2012 20:46:38, 01/01/1970 00:00:00", Owner:"isotope"
File:"hdfs://127.0.0.1:9000/user/isotope/qwanchi/MobileSampleData.txt", Modified/Accessed:"30/01/2012 20:46:38, 30/01/2012 20:46:38", Owner:"isotope"
Directory:"hdfs://127.0.0.1:9000/user/isotope/qwanchi/duplicate", Modified/Accessed:"30/01/2012 20:46:38, 01/01/1970 00:00:00", Owner:"isotope"
File:"hdfs://127.0.0.1:9000/user/isotope/qwanchi/duplicate/testdata.txt", Modified/Accessed:"30/01/2012 20:46:38, 30/01/2012 20:46:38", Owner:"isotope"
File:"hdfs://127.0.0.1:9000/user/isotope/qwanchi/testdata.txt", Modified/Accessed:"28/01/2012 20:46:38, 29/01/2012 20:46:38", Owner:"isotope"

In addition to getting directory information one can also query on a file or directory directly:

Get Path Information

hdfsSystem.SetWorkingDirectory(hdfspath);
using (HdfsFileInfoEntry pathinfo = hdfsSystem.GetPathInfo(hdfspath))
{
if (pathinfo != null)
{
string kind = pathinfo.Kind == HdfsFileInfoEntryKind.Directory ? "Directory" : "\tFile";
Console.WriteLine(string.Format(@"{0}:""{1}"", Modified/Accessed:""{2:G}, {3:G}"", Owner:""{4}""", kind, pathinfo.Name, pathinfo.LastModified, pathinfo.LastAccessed, pathinfo.Owner));
}
}

The HdfsFileSystem class also supports other operations such as copying and moving files, file renaming, deleting files, modifying security, checking a file exists, etc. The copy and move operations support copying and moving these files between systems.

So now onto creating and reading files.

Reading Files

Processing HDFS files is not that dissimilar from normal .Net file operations. Once one has opened a file for reading, operations are available for operations such as reading a byte, line, or block of bytes:

Reading Stream File Data

using (HdfsFileStream file = hdfsSystem.OpenFileStream(filename, HdfsFileAccess.Write, chunksize))
{
file.Write(dataBytes, 0, data.Length);
file.WriteByte((byte)47);
file.Flush();
}

The OpenFile operations support parameter overrides for the file block size and replication factors, whereas a value of zero implies the default values will be used.

If one wants to read the full contents of a file into a second Stream, the HdfsFileStream makes this a simple process:

Reading a File by Stream

using (HdfsFileStream hdfsStream = hdfsSystem.OpenFileStream(localhdfsfilename, HdfsFileAccess.Read))
{
using (FileStream fileStream = new FileStream(localfilestream, FileMode.Create, FileAccess.Write))
{
hdfsStream.CopyTo(fileStream);
}
}

There are other options available for reading the full contents of a file. The first option is to perform a ReadLine() until a null is returned, processed using a StreamReader:

Writing a HDFS to Local Stream

using (StreamReader reader = new StreamReader(hdfsSystem.OpenFileStream(localhdfsfilename, HdfsFileAccess.Read, chunksize)))
{
using (StreamWriter writer = new StreamWriter(localfileline, false, Encoding.UTF8))
{
string line;
while ((line = reader.ReadLine()) != null)
{
writer.WriteLine(line);
}
}
}

Alternatively, for more efficient reading of files, one can read the blocks of data into a byte array:

Reading a File in Bytes

using (HdfsFileStream file = hdfsSystem.OpenFileStream(filename, HdfsFileAccess.Read))
{
while ((chunk = file.Read(readBytes, 0, chunksize)) > 0)
{
Console.Write(Encoding.UTF8.GetString(readBytes, 0, chunk));
}
}

Other operations that are supported are PositionalReadByte(), PositionalReadBytes(), and Seek(). These operations allow reading the contents of a file from specific positions.

One final sample worth noting is copying a HDFS file to a local file using byte reads:

Writing a HDFS to Local File

using (HdfsFileStream file = hdfsSystem.OpenFileStream(localhdfsfilename, HdfsFileAccess.Read, chunksize))
{
using (FileStream stream = new FileStream(localfilewrite, FileMode.Create, FileAccess.Write))
{
while ((chunk = file.Read(readBytes, 0, chunksize)) > 0)
{
stream.Write(readBytes, 0, chunk);
}
}
}

The reason a chunk size is specified in this case is to sync the size being used for HDFS file access to the byte array used for writing the local file.

If one has a Stream reference one can also get the associated file information:

Get File Information

HdfsFileInfoEntry fileinfo = file.GetInformation();
if (fileinfo != null)
{
Console.WriteLine(string.Format(@"'{0}', Modified/Accessed:""{1:G}, {2:G}"", Owner:""{3}""", fileinfo.Name, fileinfo.LastModified, fileinfo.LastAccessed, fileinfo.Owner));
}

Also one can modify the file properties:

Modifying File Properties

file.Chown("isotope", null);
file.SetTimes(DateTime.Now.AddDays(-2), DateTime.Now.AddDays(-1));

So now onto writing files.

Writing Files

As in the case for reading, writing operations are supported for writing a byte, line, and block of bytes:

Writing File Stream Data

using (HdfsFileStream file = hdfsSystem.OpenFileStream(filename, HdfsFileAccess.Write, chunksize))
{
file.Write(dataBytes, 0, data.Length);
file.WriteByte((byte)47);
file.Flush();
}

The chunk size when opening a file is set to correspond to the size of the buffer used for writing the data.

As in the reading case, if one wants to copy a file from the local file system to an HDFS file one would write:

Writing a Local to HDFS Stream

using (HdfsFileStream file = hdfsSystem.OpenFileStream(localhdfsfilename, HdfsFileAccess.Write, chunksize))
{
using (FileStream stream = new FileStream(localfilepath, FileMode.Open, FileAccess.Read))
{
while ((chunk = stream.Read(localbytes, 0, chunksize)) > 0)
{
file.Write(localbytes, 0, chunk);
}
}
file.Flush();
}

All one has to do is read, in byte chunks, data from the local file and write the corresponding bytes to the HDFS file.

Of course one can also use the CopyTo operation:

CopyTo Local to HDFS Stream

using (HdfsFileStream file = hdfsSystem.OpenFileStream(localhdfsfilename, HdfsFileAccess.Write, chunksize))
{
using (FileStream stream = new FileStream(localfilepath, FileMode.Open, FileAccess.Read))
{
stream.CopyTo(file);
}
}

A quick word is warranted on appending to a file. Although the API currently supports open files for Append, this is only supported in Hadoop version 1.0.0 and above.

Building the Library

The download not only consists of the compiled libraries but also the full source code and sample C# application that this post is based upon. The source supports both x86 and x64 compilations. However one has to remember that if one does a 32-bit compilation then a 32-bit version of the JRE will also be required.

In the Native library the include path is defined as:

$(JAVA_HOME)\include;$(JAVA_HOME)\include\win32;$(IncludePath)

If your environment variable is not correctly defined for the Java include directory this will have to be modified.

Happy HDFS file access!

Submitting Hadoop MapReduce Jobs using PowerShell

2013-02-05T13:04:17Z

As always here is a link to the “Generics based Framework for .Net Hadoop MapReduce Job Submission” code.

In all the samples I have shown so far I have always used the command-line consoles. However this does not need to be the case, PowerShell can be used. The Console application which is used to submit the MapReduce jobs call a .Net Submissions API. As such one can call the .Net API directly from within PowerShell; as I will now demonstrate.

The key types one needs to be concerned with are:

MSDN.Hadoop.Submission.Api.SubmissionContext – The type containing the job submission options
MSDN.Hadoop.Submission.Api.SubmissionApi – The type used for submitting the job

To use the .Net API one firstly has to create the two required objects:

$SubmitterApi = $BasePath + "\Release\MSDN.Hadoop.Submission.Api.dll"
Add-Type -Path $SubmitterApi
$context = New-Object -TypeName MSDN.Hadoop.Submission.Api.SubmissionContext
$submitter = New-Object -TypeName MSDN.Hadoop.Submission.Api.SubmissionApi

After this one just has to define the context with the necessary job submission properties:

[string[]]$inputs = @("mobile/data")
[string[]]$files = @($BasePath + "\Sample\MSDN.Hadoop.MapReduceCSharp.dll")

$config = New-Object 'Tuple[string,string]'("DictionaryCapacity", "1000")
$configs = @($config)

$context.InputPaths = $inputs
$context.OutputPath = "mobile/querytimes"
$context.MapperType = "MSDN.Hadoop.MapReduceCSharp.MobilePhoneRangeMapper, MSDN.Hadoop.MapReduceCSharp"
$context.ReducerType = "MSDN.Hadoop.MapReduceCSharp.MobilePhoneRangeReducer, MSDN.Hadoop.MapReduceCSharp"
$context.Files = $files
$context.ExeConfigurations = $configs

One just has to remember that the input and files specifications are defined as string arrays.

In a recent build I added support for adding user-defined key-value pairs to the application configuration file. This ExeConfigurations property expects an array of Tuple<string, String> types, hence the object definition for the $config value.

Optionally one can also set the Data and Output format types:

$context.DataFormat = [MSDN.Hadoop.Submission.Api.DataFormat]::Text
$context.OutputFormat = [MSDN.Hadoop.Submission.Api.OutputFormat]::Text

However, this is not necessary if one is using the default Text values.

Once the context has been defined one just has to run the job:

$submitter.RunContext($context)

To call the PowerShell script from the Hadoop command-line once can use:

powershell -ExecutionPolicy unrestricted /File %BASEPATH%\SampleScripts\hadoopcstextrangesubmit.ps1

All in all a simple process.

Try F# 3.0 Released

2013-01-23T10:08:06Z

From the F# Team Blog:

If you're familiar with the site Try F#, which lets you run F# right in your browser, you'll be happy to hear that the Try F# team has launched Try F# 3.0 today! You will find updated content and tutorials, plus content that is tailored to particular areas--such as data science, numerical and scientific computing, and financial computing.

Aside from supporting all of the features of F# 3.0, Try F# 3.0 includes an enhanced user interface, support for graphs and charting, and the ability to share snippets of code with other programmers.

Try it today and don't forget to participate on the F# MSDN forum.

Hive and XML File Processing

2012-12-13T19:42:52Z

When I put together the “Generics based Framework for .Net Hadoop MapReduce Job Submission” code one of the goals was to support XML file processing. This was achieved by the creation of a modified Mahout document reader where one can specify the XML node to be presented for processing. But what if ones wants to process XML documents in Hive. Fortunately Hive similarly supports document readers, thus enabling the same document readers to be used as the basis of table definitions.

The process of enabling XML processing in Hive is relatively straightforward:

Create the table definition specifying that the input format is XML; thus exposing the necessary XML elements as columns
Parse the XML column data using xpath expressions in SELECT statements
or – Define a view on the XML table parsing out the relevant XML elements, returning them as native types

The syntax for the xpath processing in Hive can be found at: https://cwiki.apache.org/confluence/display/Hive/LanguageManual+XPathUDF

So onto a simple example.

In “SampleScripts” folder of the MapReduce Framework download there is a script that extracts Store information, in XML format, from the sample AdventureWorks database. A sample of the output is as follows:

        <Root>
  <Store>
    <BusinessEntityID>292</BusinessEntityID>
    <Name>Next-Door Bike Store</Name>
    <SalesPersonID>279</SalesPersonID>
    <Demographics>
      <StoreSurvey xmlns="http://schemas.microsoft.com/sqlserver/2004/07/adventure-works/StoreSurvey">
        <AnnualSales>800000</AnnualSales>
        <AnnualRevenue>80000</AnnualRevenue>
        <BankName>United Security</BankName>
        <BusinessType>BM</BusinessType>
        <YearOpened>1996</YearOpened>
        <Specialty>Mountain</Specialty>
        <SquareFeet>21000</SquareFeet>
        <Brands>2</Brands>
        <Internet>ISDN</Internet>
        <NumberEmployees>13</NumberEmployees>
      </StoreSurvey>
    </Demographics>
    <Modified>2008-10-13T11:15:07.497</Modified>
  </Store>
  ...
  <Store>
    <BusinessEntityID>374</BusinessEntityID>
    <Name>Immense Manufacturing Company</Name>
    <SalesPersonID>277</SalesPersonID>
    <Demographics>
      <StoreSurvey xmlns="http://schemas.microsoft.com/sqlserver/2004/07/adventure-works/StoreSurvey">
        <AnnualSales>3000000</AnnualSales>
        <AnnualRevenue>300000</AnnualRevenue>
        <BankName>Guardian Bank</BankName>
        <BusinessType>OS</BusinessType>
        <YearOpened>1998</YearOpened>
        <Specialty>Touring</Specialty>
        <SquareFeet>76000</SquareFeet>
        <Brands>4+</Brands>
        <Internet>DSL</Internet>
        <NumberEmployees>73</NumberEmployees>
      </StoreSurvey>
    </Demographics>
    <Modified>2008-10-13T11:15:07.497</Modified>
  </Store>
</Root>
  

The record reader to be used will be “XmlElementStreamingInputFormat”. This document reader using a configuration element to define the XML node to be located, which then outputs for each row a single column consisting of the complete node contents.

Using this record reader a table can be defined consisting of a single XML column:

add JARS file:///C:/Users/Carl/Projects/MSDN.Hadoop.MapReduce/Release/msdn.hadoop.readers.jar;
set xmlinput.element=Store;

CREATE EXTERNAL TABLE StoresXml (storexml string)
STORED AS INPUTFORMAT 'msdn.hadoop.mapreduce.input.XmlElementStreamingInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION '/user/Carl/stores/demographics';

The INPUTFORMAT option allows for the definition of the required document reader. The OUTPUTFORMAT specified is the default for Hive. In this example I have defined an EXTERNAL table over the directory containing the extracted XML; independently copied to the Hadoop cluster.

The ADD JARS statement ensures the document reader is available for job execution. The SET statement configures the job such that the document reader knows what XML node to extract.

Once this table is defined you can use it like any other Hive table. If one selects from this table you will get each Store element as a row. However, xpath processing allows you to extract the XML attributes as native types.

In addition to SELECT operations one also has the option of creating a VIEW that parses the XML and presents the data using native types:

CREATE VIEW Stores(BusinessEntityID, BusinessType, BankName, AnnualSales, AnnualRevenue) AS
    SELECT
        xpath_int (storexml, '/Store/BusinessEntityID'),
        xpath_string (storexml, '/Store/Demographics/*[local-name()=\'StoreSurvey\']/*[local-name()=\'BusinessType\']'),
        xpath_string (storexml, '/Store/Demographics/*[local-name()=\'StoreSurvey\']/*[local-name()=\'BankName\']'),
        xpath_double (storexml, '/Store/Demographics/*[local-name()=\'StoreSurvey\']/*[local-name()=\'AnnualSales\']'),
        xpath_double (storexml, '/Store/Demographics/*[local-name()=\'StoreSurvey\']/*[local-name()=\'AnnualRevenue\']')
    FROM StoresXml;

Using the Stores definition one can now process the XML data files through the normal Hive operations. Continuing with the same samples in the download one can now easily generate a revenue summary across the banks:

SELECT BusinessType, BankName, CAST(SUM(AnnualSales) AS INT) AS TotalSales FROM Stores
GROUP BY BusinessType, BankName;

As expected, under the covers the necessary MapReduce jobs are executed to aggregate the data, returning:

BM      Guardian Bank   43200000
BM      International Bank      43200000
BM      International Security 43200000
BM      Primary Bank & Reserve 42800000
BM      Primary International   42200000
BM      Reserve Security        42200000
BM      United Security 42200000
BS      Guardian Bank   88400000
BS      International Bank      87400000
BS      International Security 88400000
BS      Primary Bank & Reserve 88400000
BS      Primary International   87400000
BS      Reserve Security        87400000
BS      United Security 87400000
OS      Guardian Bank   192000000
OS      International Bank      186000000
OS      International Security 186000000
OS      Primary Bank & Reserve 186000000
OS      Primary International   186000000
OS      Reserve Security        186000000
OS      United Security 186000000

If you download the aforementioned code, the sample for the Hive execution can be found in the “SampleScripts” folder. The “DocumentInputReaders” folder also contains the XML document reader classes along with a usable JAR file.

Co-occurrence Approach to an Item Based Recommender Update

2012-12-05T17:30:06Z

In a previous post I talked about a Co-occurrence Approach to an Item Based Recommender, that utilized the Math.Net Numerics library. Recently the Math.Net Numerics library was updated to version 2.3.0. With this version of the library I was able to update the code to more efficiently read the Sparse Matrix entries. As such I have updated the code to reflect these library changes:

http://code.msdn.microsoft.com/Co-occurrence-Approach-to-57027db7

The new Mat.Net Numerics Library changes were around the storage of the Vector and Matrix elements. As such I was now able to access the storage directly and use the Compress Sparse Row Matrix format to more efficiently access the Sparse Matrix elements.

The original code that accessed the elements of the Sparse Matrix was a simple row/column traverse:

let getQueue (products:int array) =         
     // Define the priority queue and lookup table
     let queue = PriorityQueue(coMatrix.ColumnCount)
     let lookup = HashSet(products)
 
     // Add the items into a priority queue
     products
     |> Array.iter (fun item ->
         let itemIdx = item - offset
         if itemIdx >= 0 && itemIdx < coMatrix.ColumnCount then
             seq { 
                 for idx = 0 to (coMatrix.ColumnCount - 1) do
                     let productIdx = idx + offset
                     let item = coMatrix.[itemIdx, idx]
                     if (not (lookup.Contains(productIdx))) && (item > 0.0) then
                         yield KeyValuePair(item, productIdx)
             }
             |> queue.Merge)
     // Return the queue
     queue

Now one has access to the storage elements I was able to more efficiently access just the sparse element values:

products
 |> Array.iter (fun item ->
     let itemIdx = item - offset
     let sparse = coMatrix.Storage :?> SparseCompressedRowMatrixStorage<double>
     let last = sparse.RowPointers.Length - 1 
     
     if itemIdx >= 0 && itemIdx <= last then
         let (startI, endI) =
             if itemIdx = last then
                 (sparse.RowPointers.[itemIdx], sparse.RowPointers.[itemIdx])
             else
                 (sparse.RowPointers.[itemIdx], sparse.RowPointers.[itemIdx + 1] - 1)
         seq { 
             for idx = startI to endI do
                 let productIdx = sparse.ColumnIndices.[idx] + offset
                 let item = sparse.Values.[idx]
                 if (not (lookup.Contains(productIdx))) && (item > 0.0) then
                     yield KeyValuePair(item, productIdx)
         }
         |> queue.Merge)
 // Return the queue
 queue

In the new version of the code The Values array provides access to the underlying non-empty values. The RowPointers array provides access to the value indexes where each row starts. Finally, the ColumnIndicies are the column indices corresponding to the values.

Other than this change all other aspects of the library’s usage were effectively unchanged; including the MapReduce code (postings can be found here), as this uses a collection of Vector types. I did however update the job submission scripts.

Implementing a MapReduce Join with Hadoop and the .Net Framework

2012-11-13T20:46:10Z

I have often been asked how does one implement a Join whilst writing MapReduce code. As such, I thought it would be useful to add an additional sample demonstrating how this is achieved. There are multiple mechanisms one can employ to perform a Join operation, and the one to be discussed will be a Reduce Side 1-to-many join.

As always this sample, amongst others, can be found in the “Generics based Framework for .Net Hadoop MapReduce Job Submission” code download; within the Samples folder.

Join Semantics

Say one wants to join two sets of data, A and B, via a common set attribute. Set A could be defined as a collection of tuples of the form (k_i, a_i, A_i), where k represents the key value on which we want to do the join, s the set item unique identifier, and S the other attributes. For this set each k value would correspond to a unique value of a.

Set B would be similarly defined as a collection of tuples of the form (k_i, b_x, B_x). In this case each value of k would equate to multiple values of b.

As an example, set A could be thus represented by an OrderHeader type and B the corresponding OrderDetail types. The k and a values in this case would represent the sales order identifier. The b value would represent the sales order detail identifier.

The basic concept is that the MapReduce job will create a new set C, which would be defined by the tuple collection (k_i, a_x, b_y, A_x, B_y). For each key identifier k, there would be a single value of a, and multiple values for b.

(k₆₄, a₅₂, b₁₀₆, A₅₂, B₁₀₆)
(k₆₄, a₅₂, b₁₂₁, A₅₂, B₁₂₁)
…
(k₆₄, a₅₂, b₂₃₄, A₅₂, B₂₃₄)

Furthering the sample this new set C could be represented by a new type being the aggregate of the OrderHeader and OrderDetail types; say OrderLine. The SalesOrderId would be the common key attribute.

To perform the Reduce Side join the Mapper would read in the data representing both the sets of data. When processing set A data the emitted value would be (k_i, a_x, A_x, Ø, Ø), and for set B the value (k_i, Ø, Ø, b_y, B_y). In both cases the emitted key would be common key attribute value.

The Reducer would receive the set values for each shared key attribute; namely a single (k_i, a_x, A_x, Ø, Ø) value, and multiple (k_i, Ø, Ø, b_y, B_y) values. It would then emit the full set of (k_i, a_x, b_y, A_x, B_y) values; for each key attribute.

As an example when processing orders the input data would be parsed into either an OrderHeader or OrderDetail object. In both cases the Mapper would emit an OrderLine item; with a null value for the missing data. The Reducer would then read these values for a specific SalesOrderID key value and then emit the set of complete OrderLine values.

So as an example, lets do a join between order header and detail information, from the SQL Server AdventureWorks sample database. The sample download also includes a BCP scripts to extract sales data from the sample database. For completeness I have also written the sample in both C# (with heavy use of LINQ) and F#.

Defining the Types

To start we need to define types that represent the sales header and detail information. These type definitions, listed below, are OrderHeader and OrderDetail. As mentioned we also need an aggregate type, called OrderLine, that aggregates both the header and detail types.

C# Classes

 /// Order Header Information
public class OrderHeader
{
    public int SalesOrderID { get; set; }
    public DateTime OrderDate { get; set; }
    public DateTime ShipDate { get; set; }
    public int CustomerID { get; set; }
    public string SalesOrderNumber { get; set; }
    public string PurchaseOrderNumber { get; set; }
    public string AccountNumber { get; set; }
}
 
/// Order Header Information
public class OrderDetail
{
    public int SalesOrderID { get; set; }
    public int SalesOrderDetailID { get; set; }
    public int ProductID { get; set; }
    public decimal OrderQty { get; set; }
    public decimal UnitPrice { get; set; }
}
 
/// Order Line Information
public class OrderLine
{
    public OrderHeader OrderHeader { get; set; }
    public OrderDetail OrderDetail { get; set; }
}
 

F# Records

 /// Order Header Information
type OrderHeader = { SalesOrderID:int; OrderDate:DateTime; ShipDate:DateTime; CustomerID:int; SalesOrderNumber:string; PurchaseOrderNumber:string; AccountNumber:string }
 
/// Order Detail Information
type OrderDetail = { SalesOrderID:int; SalesOrderDetailID:int; ProductID:int; OrderQty:decimal; UnitPrice:decimal }
 
/// Order Line Information
type OrderLine = { OrderHeader:OrderHeader; OrderDetail:OrderDetail }
 

The OrderLine aggregated type is used as the output from the Mapper and the Reducer.

Mapper Processing

The purpose of the mapper is to read in both the header and detail data. For each data record the mapper outputs a value of an OrderLine instance, along with a key value of the SalesOrderID value. When processing a sales header data item an OrderLine is output with a null value for the OrderDetail; and of course vice-versa when processing a detail data item.

The code for parsing the data items is as below. In this instance I have made the determination of the data item type by inspecting the second data value. In this case I am looking for an OrderDate; however one can adopt many different approaches based on the input data.

C# Mapper

 // Processes the Order Header and Detail files (In Memory)
class OrderJoinMemoryMapper : MapperBaseText<OrderLine>
{
    public override IEnumerable<Tuple<string, OrderLine>> Map(string value)
    {
        var splits = value.Split('\t');
        int salesOrderID = Int32.Parse(splits[0]);
 
        if (splits[1].Contains("-"))
        {
            DateTime orderDate = DateTime.Parse(splits[1]);
            DateTime shipDate = DateTime.Parse(splits[2]);
            int customerID = Int32.Parse(splits[3]);
            string salesOrderNumber = splits[4];
            string purchaseOrderNumber = splits[5];
            string accountNumber = splits[6];
            var header = new OrderHeader()
            {
                SalesOrderID = salesOrderID,
                OrderDate = orderDate,
                ShipDate = shipDate,
                CustomerID = customerID,
                SalesOrderNumber = salesOrderNumber,
                PurchaseOrderNumber = purchaseOrderNumber,
                AccountNumber = accountNumber
            };
            yield return Tuple.Create(splits[0], new OrderLine() { OrderHeader = header, OrderDetail = null });
        }
        else
        {
            int salesOrderDetailID = Int32.Parse(splits[1]);
            int productID = Int32.Parse(splits[2]);
            decimal orderQty = Decimal.Parse(splits[3]);
            decimal unitPrice = Decimal.Parse(splits[4]);
            var detail = new OrderDetail()
            {
                SalesOrderID = salesOrderID,
                SalesOrderDetailID = salesOrderDetailID,
                ProductID = productID,
                OrderQty = orderQty,
                UnitPrice = unitPrice
            };
            yield return Tuple.Create(splits[0], new OrderLine() { OrderHeader = null, OrderDetail = detail });
        }
    }
}
 

F# Mapper

 // Processes the Order Header and Detail files (In Memory)
type OrderJoinMemoryMapper() =
    inherit MapperBaseText<OrderLine>()
 
    // Performs the split into the correct type
    let (|Header|Detail|Unknown|) (value:string) =
        try
            let splits = value.Split('\t')
            let salesOrderID = Int32.Parse(splits.[0])
            if splits.[1].Contains("-") then
                // Processing a Header record
                let orderDate = DateTime.Parse(splits.[1])
                let shipDate = DateTime.Parse(splits.[2])
                let customerID = Int32.Parse(splits.[3])
                let salesOrderNumber = splits.[4]
                let purchaseOrderNumber = splits.[5]
                let accountNumber = splits.[6]
                Header (splits.[0], { OrderHeader.SalesOrderID = salesOrderID; OrderDate = orderDate; ShipDate = shipDate; CustomerID = customerID;
                                    SalesOrderNumber = salesOrderNumber; PurchaseOrderNumber = purchaseOrderNumber; AccountNumber = accountNumber })
            else
                // Processing a detail record
                let salesOrderDetailID = Int32.Parse(splits.[1])
                let productID = Int32.Parse(splits.[2])
                let orderQty = Decimal.Parse(splits.[3])
                let unitPrice = Decimal.Parse(splits.[4])
                Detail (splits.[0], {OrderDetail.SalesOrderID = salesOrderID; SalesOrderDetailID = salesOrderDetailID; ProductID = productID;
                                    OrderQty = orderQty; UnitPrice = unitPrice })
        with
        | :? System.ArgumentException -> Unknown
 
    // Map the data from input name/value to output name/value
    override self.Map (value:string) =
        seq {
            match value with
            | Header (key, header) -> yield (key, { OrderLine.OrderHeader = header; OrderDetail = Unchecked.defaultof<OrderDetail> })
            | Detail (key, detail) -> yield (key, { OrderLine.OrderHeader = Unchecked.defaultof<OrderHeader>; OrderDetail = detail })
            | Unknown -> ()
        }
 

Literally that is it for the Mapper. It determines the type of input line and emits an OrderLine item.

As you can see the key is the SalesOrderID value. More on this later, when talking about a performance optimization.

Reducer Processing

The purpose of the Reducer is to locate the OrderHeader value for each SalesOrderID key value, and for each corresponding OrderDetail value output a complete OrderLine; consisting of the located header value and the detail values.

In this version of the code, again shown below, use is made of a List of OrderDetail items. Once all the values have been read then the OrderLine sequence of values are returned.

C# Reducer

 // Performs the combined data 1-many join (In Memory)
class OrderJoinMemoryReducer : ReducerBase<OrderLine, OrderLine>
{
    public override IEnumerable<Tuple<string, OrderLine>> Reduce(string key, IEnumerable<OrderLine> values)
    {
        OrderHeader orderHeader = null;
        var orderDetails = new List<OrderDetail>();
 
        foreach (var line in values)
        {
            if (line.OrderDetail != null)
            {
                orderDetails.Add(line.OrderDetail);
            }
            else
            {
                orderHeader = line.OrderHeader;
            }
        }
 
        if (orderHeader != null)
        {
            return orderDetails.Select(detail => Tuple.Create(key, new OrderLine() { OrderHeader = orderHeader, OrderDetail = detail }));
        }
        else
        {
            return Enumerable.Empty<Tuple<string, OrderLine>>();
        }
    }
}
 

F# Reducer

 // Performs the combined data 1-many join (In Memory)
type OrderJoinMemoryReducer() =
    inherit ReducerBase<OrderLine, OrderLine>()
 
    override self.Reduce (key:string) (values:seq<OrderLine>) =
        let orderHeader = ref Unchecked.defaultof<OrderHeader>
        let hasValue value = not (obj.ReferenceEquals (value, Unchecked.defaultof<_>))
 
        let orderDetails =
            values
            |> Seq.choose (fun item ->
                if (hasValue item.OrderDetail) then
                    Some(item.OrderDetail)
                else
                    orderHeader := item.OrderHeader
                    None)
 
        if (hasValue orderHeader) then
            orderDetails
            |> Seq.toList
            |> Seq.map (fun item ->
                (key, { OrderLine.OrderHeader = !orderHeader; OrderDetail = item }))
        else
            Seq.empty
 

As you can see this code has a memory limitation. For each SalesOrderID value all the corresponding detail values need to be cached in memory, as no determination can be made as to when the header value is located. This may not be an issue with a small number of details associated to a header, but what about data where this number can be extremely large.

This limitation however is easily overcome through the use of a secondary sort. Basically two key values are used to ensure the header value always arrives first within the Reducer.

Secondary Sort Optimization

When processing the order data for a particular key within the reducer, if one knows that the first value will always be the header data then this can be saved for detail data processing. Thus when reading the subsequent detail lines the aggregated OrderLine instance values can be emitted directly without the intermediate List processing step. This is where a secondary sort comes into play.

If one uses two keys rather than one, it is possible to sort the data such that the first value, for each SalesOrderID, will be the header data. This can be achieved in our case by using a secondary sort key of the SalesOrderDetailID. For the header one can then just use a value of zero to ensure it is the first value; as all details items have a positive value. Of course, different approaches can be taken depending on the data domain.

Although the data is sorted on two separate keys, the partitioning must be such that the data for each SalesOrderID is sent to a single Reducer; spanning multiple SalesOrderDetailID values.

Using this approach here is the code for the modified Mapper logic:

C# Mapper

 // Processes the Order Header and Detail files
class OrderJoinMapper : MapperBaseText<OrderLine>
{
    public override IEnumerable<Tuple<string, OrderLine>> Map(string value)
    {
        var splits = value.Split('\t');
        int salesOrderID = Int32.Parse(splits[0]);
 
        if (splits[1].Contains("-"))
        {
            DateTime orderDate = DateTime.Parse(splits[1]);
            DateTime shipDate = DateTime.Parse(splits[2]);
            int customerID = Int32.Parse(splits[3]);
            string salesOrderNumber = splits[4];
            string purchaseOrderNumber = splits[5];
            string accountNumber = splits[6];
            string key = Context.FormatKeys(splits[0], "0");
            var header = new OrderHeader()
            {
                SalesOrderID = salesOrderID,
                OrderDate = orderDate,
                ShipDate = shipDate,
                CustomerID = customerID,
                SalesOrderNumber = salesOrderNumber,
                PurchaseOrderNumber = purchaseOrderNumber,
                AccountNumber = accountNumber
            };
            yield return Tuple.Create(key, new OrderLine() { OrderHeader = header, OrderDetail = null });
        }
        else
        {
            int salesOrderDetailID = Int32.Parse(splits[1]);
            int productID = Int32.Parse(splits[2]);
            decimal orderQty = Decimal.Parse(splits[3]);
            decimal unitPrice = Decimal.Parse(splits[4]);
            string key = Context.FormatKeys(splits[0], splits[1]);
            var detail = new OrderDetail()
            {
                SalesOrderID = salesOrderID,
                SalesOrderDetailID = salesOrderDetailID,
                ProductID = productID,
                OrderQty = orderQty,
                UnitPrice = unitPrice
            };
            yield return Tuple.Create(key, new OrderLine() { OrderHeader = null, OrderDetail = detail });
        }
    }
}
 

F# Mapper

 // Processes the Order Header and Detail files
type OrderJoinMapper() =
    inherit MapperBaseText<OrderLine>()
 
    // Performs the split into the correct type
    let (|Header|Detail|Unknown|) (value:string) =
        try
            let splits = value.Split('\t')
            let salesOrderID = Int32.Parse(splits.[0])
            if splits.[1].Contains("-") then
                // Processing a Header record
                let orderDate = DateTime.Parse(splits.[1])
      ��         let shipDate = DateTime.Parse(splits.[2])
                let customerID = Int32.Parse(splits.[3])
                let salesOrderNumber = splits.[4]
                let purchaseOrderNumber = splits.[5]
                let accountNumber = splits.[6]
                let key = Context.FormatKeys(splits.[0], "0")
                Header (key, { OrderHeader.SalesOrderID = salesOrderID; OrderDate = orderDate; ShipDate = shipDate; CustomerID = customerID;
                                    SalesOrderNumber = salesOrderNumber; PurchaseOrderNumber = purchaseOrderNumber; AccountNumber = accountNumber })
            else
                // Processing a detail record
                let salesOrderDetailID = Int32.Parse(splits.[1])
                let productID = Int32.Parse(splits.[2])
                let orderQty = Decimal.Parse(splits.[3])
                let unitPrice = Decimal.Parse(splits.[4])
                let key = Context.FormatKeys(splits.[0], splits.[1])
                Detail (key, {OrderDetail.SalesOrderID = salesOrderID; SalesOrderDetailID = salesOrderDetailID; ProductID = productID;
                                    OrderQty = orderQty; UnitPrice = unitPrice })
        with
        | :? System.ArgumentException -> Unknown
 
    // Map the data from input name/value to output name/value
    override self.Map (value:string) =
        seq {
            match value with
            | Header (key, header) -> yield (key, { OrderLine.OrderHeader = header; OrderDetail = Unchecked.defaultof<OrderDetail> })
            | Detail (key, detail) -> yield (key, { OrderLine.OrderHeader = Unchecked.defaultof<OrderHeader>; OrderDetail = detail })
            | Unknown -> ()
        }
 

The multiple keys values are delimitated with a Tab value.

The Reducer code then becomes a lot simpler as no List caching is involved; and where LINQ enables succinct C# code.

C# Reducer

 // Performs the combined data 1-many join
class OrderJoinReducer : ReducerBase<OrderLine, OrderLine>
{
    public override IEnumerable<Tuple<string, OrderLine>> Reduce(string key, IEnumerable<OrderLine> values)
    {
        string salesOrderID = Context.GetKeys(key)[0];
        OrderHeader orderHeader = values.ElementAt(0).OrderHeader;
 
        return values.Skip(1).Select(detail => Tuple.Create(salesOrderID, new OrderLine() { OrderHeader = orderHeader, OrderDetail = detail.OrderDetail }));
    }
}
 

F# Reducer

 // Performs the combined data 1-many join
type OrderJoinReducer() =
    inherit ReducerBase<OrderLine, OrderLine>()
 
    override self.Reduce (key:string) (values:seq<OrderLine>) =
        
        let salesOrderID = Context.GetKeys(key).[0]
        let orderHeader = Seq.nth 0 values
        
        values
        |> Seq.skip 1
        |> Seq.map (fun item ->
            (salesOrderID, { OrderLine.OrderHeader = orderHeader.OrderHeader; OrderDetail = item.OrderDetail }))
 

As you can see the code is probably simpler, especially when it comes to the Reducer, and is also more efficient than the In-Memory approach.

Submitting the Jobs

For the In-Memory code the job submission is quite simple:

%BASEPATH%\MSDN.Hadoop.MapReduce\Release\MSDN.Hadoop.Submission.Console.exe
-input "join/data"
-output "join/order"
-mapper "MSDN.Hadoop.MapReduceFSharp.OrderJoinMemoryMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.OrderJoinMemoryReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%BASEPATH%\MSDN.Hadoop.MapReduceFSharp\Release\MSDN.Hadoop.MapReduceFSharp.dll"

However, for the optimized version one needs to accommodate the fact that the sorting takes place on two keys, and the data partitioning only on one. This is achieved by setting these exact options:

%BASEPATH%\MSDN.Hadoop.MapReduce\Release\MSDN.Hadoop.Submission.Console.exe
-input "join/data"
-output "join/order"
-mapper "MSDN.Hadoop.MapReduceFSharp.OrderJoinMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.OrderJoinReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%BASEPATH%\MSDN.Hadoop.MapReduceFSharp\Release\MSDN.Hadoop.MapReduceFSharp.dll"
-numberKeys 2 -numberPartitionKeys 1

The reason these options were added to version 1.0.0 of the code was to accommodate this exact type of processing.

Conclusion

To conclude here is some sample output from the join.

43667    {"OrderDetail":{"OrderQty":3,"ProductID":710,"SalesOrderDetailID":77,"SalesOrderID":43667,"UnitPrice":5.7000},"OrderHeader":{"AccountNumber":"10-4020-000646","CustomerID":29974,"OrderDate":"\/Date(1120172400000+0100)\/","PurchaseOrderNumber":"PO15428132599","SalesOrderID":43667,"SalesOrderNumber":"SO43667","ShipDate":"\/Date(1120777200000+0100)\/"}}
 43667    {"OrderDetail":{"OrderQty":1,"ProductID":773,"SalesOrderDetailID":78,"SalesOrderID":43667,"UnitPrice":2039.9940},"OrderHeader":{"AccountNumber":"10-4020-000646","CustomerID":29974,"OrderDate":"\/Date(1120172400000+0100)\/","PurchaseOrderNumber":"PO15428132599","SalesOrderID":43667,"SalesOrderNumber":"SO43667","ShipDate":"\/Date(1120777200000+0100)\/"}}
 43667    {"OrderDetail":{"OrderQty":1,"ProductID":778,"SalesOrderDetailID":79,"SalesOrderID":43667,"UnitPrice":2024.9940},"OrderHeader":{"AccountNumber":"10-4020-000646","CustomerID":29974,"OrderDate":"\/Date(1120172400000+0100)\/","PurchaseOrderNumber":"PO15428132599","SalesOrderID":43667,"SalesOrderNumber":"SO43667","ShipDate":"\/Date(1120777200000+0100)\/"}}
 43667    {"OrderDetail":{"OrderQty":1,"ProductID":775,"SalesOrderDetailID":80,"SalesOrderID":43667,"UnitPrice":2024.9940},"OrderHeader":{"AccountNumber":"10-4020-000646","CustomerID":29974,"OrderDate":"\/Date(1120172400000+0100)\/","PurchaseOrderNumber":"PO15428132599","SalesOrderID":43667,"SalesOrderNumber":"SO43667","ShipDate":"\/Date(1120777200000+0100)\/"}}

Utilizing the output is merely a case of serializing the data back into the OrderLine type.

Hopefully I have demonstrated that MapReduce joins are not that difficult to perform. With a little thought around the types, the code is not much more than data parsing.

A word of caution, compared to standard database joins, MapReduce joins are slow. If you can join the data before running MapReduce processing then this should be your preferred approach.

Framework for .Net Hadoop MapReduce Job Submission V1.0 Release

2012-10-10T10:31:00Z

It has been a few months since I have made a change to the “Generics based Framework for .Net Hadoop MapReduce Job Submission” code. However I was going to put together a sample for a Reduce side join and came across a issue around the usage of partitioners. As such I decided to add support for custom partitioners and comparators, before stamping the release as a version 1.0.

The submission options to be added will be -numberPartitionKeys (for partitioning the data) and -comparatorOption (for sorting). Before talking about these options lets cover a little of the Hadoop Streaming Documentation.

A Useful Partitioner Class

Hadoop has a library class, KeyFieldBasedPartitioner, that allows the MapReduce framework to partition the map outputs on key segments rather than the whole key. Consider the following job submission options:

    -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner 
    -Dstream.num.map.output.key.fields=4 
    -Dmapred.text.key.partitioner.options=-k1,2

The -Dstream.num.map.output.key.fields=4 option specifies that the map output keys will have 4 fields. In this case the the -Dmapred.text.key.partitioner.options=-k1,2 option tells the MapReduce framework to partition the map outputs by the first two fields of the key; rather than the full set of four keys.This guarantees that all the key/value pairs with the same first two fields in the keys will be partitioned to the same reducer.

To simplify this specification the submission framework supports the following command line options:

    -numberKeys 4
    -numberPartitionKeys 2

When writing Streaming applications one normally has to do the partition processing in the Reducer. However with these options the framework will correctly send the appropriate data to the Reducer. This was the biggest change that needed to be made to the Framework.

This partition processing is important for handling Reduce side joins; the topic of my next blog post.

A Useful Comparator Class

Hadoop also has a library class, KeyFieldBasedComparator, that provides sorting the key data. Consider the following job submission options:

    -D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator
    -D mapred.text.key.comparator.options=-k2,2nr

Here the MapReduce framework will sort the outputs on the second key field; the -Dmapred.text.key.comparator.options=-k2,2nr option. The -n specifies that the sorting is numerical and the -r specifies that the result should be reversed. The sort options are similar to a Unix sort, but here are some simple examples:

Reverse ordering: -Dmapred.text.key.comparator.options=-r

Numeric ordering: -Dmapred.text.key.comparator.options=-n

Specific sort specification: -Dmapred.text.key.comparator.options=-kx,y

For the general sort specification the -k flag is used, which allows you to specify the sorting key using the the x, y values; as in the sample above.

To simplify this specification the submission framework supports the following command line option:

    -comparatorOption ”-k2,2nr”

The framework then takes care of setting the appropriate job configuration values.

Summary

Although one could define the Partitioner and Comparator options using the job configuration parameters hopefully these new options make the process a lot simpler. In the case of the Partitioner options it also allows the framework to easily identity the difference in the number of sorting and partitioner keys. This allows the correct data to be sent to each Reducer.

As mentioned, using these options, in my next post I will cover how to use the Framework for doing a Reduce side join.

Execution Time Based Heuristic Custom Task Scheduler

2012-10-02T18:36:01Z

If you follow the samples for Parallel Programming with the .Net Framework, you may have come across the ParallelExtensionsExtras and the Additional TaskSchedulers. Although these samples cover a broad set of requirements I recently came across another that could be satisfied with the creation of a new custom task scheduler.

In the current samples there is a custom task scheduler that supports a Maximum Degree of Parallelism (MDOP). However, what if one wanted to adjust the MDOP based on the task execution times.

As an example consider calling a web service, or performing any data processing, from a message queue. I recently came across this requirement, where if the response times drop then the MDOP also needed to drop. One could take the approach of a service that manages its threads but a custom task scheduler enables a far greater set of usage scenarios.

The full code can be located here: Execution Time Based Heuristic Custom Task Scheduler

LimitedConcurrencyLevelTaskScheduler Class

As a starting point here is the code from the PfX Team covering a custom task scheduler that allows one to specify a fixed MDOP:

 using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
 
namespace MSDN.Schedulers
{
    /// <summary>Provides a task scheduler that ensures a maximum concurrency level while running on top of the ThreadPool.</summary>
    public sealed class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
    {
        /// <summary>Values defining the current state of the scheduler.</summary>
        public struct ExecutionState
        {
            /// <summary>Current level of concurrency.</summary>
            public int CurrentConcurrency { get; internal set; }
 
            /// <summary>Current count of awaiting task.</summary>
            public int AwaitingTasks { get; internal set; }
        }
 
        /// <summary>Whether the current thread is processing work items.</summary>
        [ThreadStatic]
        private static bool CurrentThreadIsProcessingItems;
 
        /// <summary>The list of tasks to be executed.</summary>
        private readonly LinkedList<Task> tasks = new LinkedList<Task>(); // protected by lock(tasks)
 
        /// <summary>The maximum concurrency level allowed by this scheduler.</summary>
        private readonly int maxDegreeOfParallelism;
 
        /// <summary>Whether the scheduler is currently processing work items.</summary>
        private int delegatesQueuedOrRunning = 0; // protected by lock(_tasks)
 
        /// <summary>Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the specified degree of parallelism.</summary>
        /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
        public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
        {
            if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
            this.maxDegreeOfParallelism = maxDegreeOfParallelism;
        }
 
        /// <summary>Queues a task to the scheduler.</summary>
        /// <param name="task">The task to be queued.</param>
        protected override void QueueTask(Task task)
        {
            // Add the task to the list of tasks to be processed.  If there aren't enough
            // delegates currently queued or running to process tasks, schedule another.
            lock (this.tasks)
            {
                this.tasks.AddLast(task);
                if (this.delegatesQueuedOrRunning < this.maxDegreeOfParallelism)
                {
                    ++this.delegatesQueuedOrRunning;
                    NotifyThreadPoolOfPendingWork();
                }
            }
        }
 
        /// <summary>Informs the ThreadPool that there's work to be executed for this scheduler.</summary>
        private void NotifyThreadPoolOfPendingWork()
        {
            ThreadPool.UnsafeQueueUserWorkItem(_ =>
            {
                // Note that the current thread is now processing work items.
                // This is necessary to enable inlining of tasks into this thread.
                CurrentThreadIsProcessingItems = true;
                try
                {
                    // Process all available items in the queue.
                    while (true)
                    {
                        Task item;
                        lock (this.tasks)
                        {
                            // When there are no more items to be processed,
                            // note that we're done processing, and get out.
                            if (this.tasks.Count == 0)
                            {
                                --this.delegatesQueuedOrRunning;
                                break;
                            }
 
                            // Get the next item from the queue
                            item = this.tasks.First.Value;
                            this.tasks.RemoveFirst();
                        }
 
                        // Execute the task we pulled out of the queue
                        base.TryExecuteTask(item);
                    }
                }
                // We're done processing items on the current thread
                finally
                {
                    CurrentThreadIsProcessingItems = false;
                }
            }, null);
        }
 
        /// <summary>Attempts to execute the specified task on the current thread.</summary>
        /// <param name="task">The task to be executed.</param>
        /// <param name="taskWasPreviouslyQueued">Indicator for whether the task was previously dequeued.</param>
        /// <returns>Whether the task could be executed on the current thread.</returns>
        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        {
            // If this thread isn't already processing a task, we don't support inlining
            if (!CurrentThreadIsProcessingItems) return false;
 
            // If the task was previously queued, remove it from the queue
            if (taskWasPreviouslyQueued) TryDequeue(task);
 
            // Try to run the task.
            return base.TryExecuteTask(task);
        }
 
        /// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
        /// <param name="task">The task to be removed.</param>
        /// <returns>Whether the task could be found and removed.</returns>
        protected override bool TryDequeue(Task task)
        {
            lock (this.tasks)
            {
                return this.tasks.Remove(task);
            }
        }
 
        /// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
        /// <returns>An enumerable of the tasks currently scheduled.</returns>
        protected override IEnumerable<Task> GetScheduledTasks()
        {
            bool lockTaken = false;
 
            try
            {
                Monitor.TryEnter(this.tasks, ref lockTaken);
                if (lockTaken) return this.tasks.ToArray();
                else throw new NotSupportedException();
            }
            finally
            {
                if (lockTaken) Monitor.Exit(this.tasks);
            }
        }
 
        /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
        public override int MaximumConcurrencyLevel
        {
            get
            {
                return this.maxDegreeOfParallelism;
            }
        }
 
        /// <summary>Gets the current Execution State of the scheduler.</summary>
        public ExecutionState SchedulerExecutionState
        {
            get
            {
                ExecutionState state = new ExecutionState();
                state.CurrentConcurrency = this.delegatesQueuedOrRunning;
                state.AwaitingTasks = this.tasks.Count;
 
                return state;
            }
        }
 
        /// <summary>Gets the current Processing state of the scheduler.</summary>
        public bool IsProcessing
        {
            get
            {
                return (this.tasks.Count > 0 || this.delegatesQueuedOrRunning > 0);
            }
        }
    }
}
 

To use the task scheduler one just has to create a task scheduler instance, with the required MDOP, and then create a Factory with the newly created scheduler:

LimitedConcurrencyLevelTaskScheduler lclts = new LimitedConcurrencyLevelTaskScheduler(5);
 TaskFactory factory = new TaskFactory(lclts);
 
 factory.StartNew(() =>
     {
         for (int i = 0; i < 10; i++)
         {
             Console.WriteLine("{0} on thread {1}", i, Thread.CurrentThread.ManagedThreadId);
         }
     }
 );

Alternatively to use the scheduler in Parallel loops, one will again create the task scheduler and use this when defining the ParallelOptions class for the parallel loop:

LimitedConcurrencyLevelTaskScheduler lclts = new LimitedConcurrencyLevelTaskScheduler(5);
 ParallelOptions options = new ParallelOptions();
 options.TaskScheduler = lclts;
 
 Parallel.For(
         0,
         50,
         options,
         (i) =>
         {                        
             Thread.Sleep(100);
             Console.WriteLine("Finish Thread={0}, i={1}", Thread.CurrentThread.ManagedThreadId, i);
         }
     );

All easy stuff. So onto the new task scheduler.

ResponseLimitedConcurrencyTaskScheduler Class

In meeting the requirement to vary the MDOP based on execution times there are three main differences one needs to accommodate. Firstly, one needs a function that can be used for determining the MDOP based on average task execution times. Secondly, one needs to record the actual task execution times in such a manner that the average can be determined and passed to the MDOP calculator function. Lastly, the MDOP needs to be re-evaluated, and the number of executing threads adjusted accordingly.

To support the requirement for determining the MDOP the scheduler class constructor now supports the following prototype:

public ResponseLimitedConcurrencyTaskScheduler(Func<int, int> funcMdop,
    ResponseLimitedConcurrencyTaskConfiguration configuration)

This MDOP function is designed to accept an average execution time and return the desired MDOP. To calculate the average execution times a ConcurrentStack<int> is used to record the task execution times. On a timed schedule the top configured items from the stack are averaged, with this value being used in the MDOP calculation.

Once the MDOP has been calculated it is used in the NotifyThreadPoolOfPendingWork() call to ensure the correct MDOP is used when processing threads.

So without further adieu here is the full code listing for the scheduler:

namespace MSDN.Schedulers
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
 
    /// <summary>Provides a task scheduler that ensures a maximum concurrency level while running on top of the ThreadPool.</summary>
    public sealed class ResponseLimitedConcurrencyTaskScheduler : TaskScheduler, IDisposable
    {
        /// <summary>Values defining the current state of the scheduler.</summary>
        public struct ExecutionState
        {
            /// <summary>Gets the current Max Degree of Parallelism.</summary>
            public int MaxDegreeOfParallelism { get; internal set; }
 
            /// <summary>Gets the current level of concurrency.</summary>
            public int CurrentConcurrency { get; internal set; }
 
            /// <summary>Gets the current count of awaiting task.</summary>
            public int AwaitingTasks { get; internal set; }
 
            /// <summary>Gets the last sampled average of the execution times.</summary>
            public int LastSampledAverage { get; internal set; }
        }
 
        /// <summary>Whether the current thread is processing work items.</summary>
        [ThreadStatic]
        private static bool CurrentThreadIsProcessingItems;
 
        /// <summary>Scheduler running configuration.</summary>
        private readonly ResponseLimitedConcurrencyTaskConfiguration configuration;
 
        /// <summary>The list of tasks to be executed.</summary>
        private readonly LinkedList<Task> tasks = new LinkedList<Task>(); // protected by lock(_tasks)
 
        /// <summary>The maximum concurrency level allowed by this scheduler for the current response times.</summary>
        private volatile int maxDegreeOfParallelism;
 
        /// <summary>Last sampled average of the execution times.</summary>
        private volatile int lastSampledAverage;
 
        /// <summary>Whether the scheduler is currently processing work items.</summary>
        private int delegatesQueuedOrRunning = 0; // protected by lock(_tasks)
 
        /// <summary>Stack for holding execution times for calculating MDOP </summary>
        private ConcurrentStack<int> timings = new ConcurrentStack<int>();
 
        /// <summary>Object to ensure MDOP is only set on a single thread.</summary>
        private object timingsSync = new object();
 
        /// <summary>Function determining the MDOP, where a zero average gives the starting MDOP.</summary>
        private Func<int, int> funcMdop;
 
        /// <summary>The timer for refreshing the scheduler MDOP.</summary>
        private Timer refreshTimer;
 
        /// <summary>Indicator for whether Dispose has been called.</summary>
        private bool disposed = false;
 
        /// <summary> Initializes a new instance of the ResponseLimitedConcurrencyTaskScheduler class with the specified MDOP function. </summary>
        /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism function provider for this scheduler.</param>
        /// <param name="configuration">Scheduler configuration.</param>
        public ResponseLimitedConcurrencyTaskScheduler(Func<int, int> funcMdop, ResponseLimitedConcurrencyTaskConfiguration configuration)
        {
            // Finally ensure zero average gives a starting MDOP
            if (funcMdop(0) < 0) { throw new ArgumentOutOfRangeException("funcMdop", "Calculated Zero Max Degree of Parallelism cannot be negative."); }
 
            this.funcMdop = funcMdop;
            this.configuration = configuration;
 
            // Get the default MDOP
            int mdop = this.funcMdop(0);
            this.maxDegreeOfParallelism = mdop == 0 ? this.configuration.DefaultDegreeOfParallelism : mdop;
            this.lastSampledAverage = 0;
 
            // Setup the refresh of the MDOP
            int rate = this.configuration.SamplingRateMilliseconds;
            this.refreshTimer = new Timer(state => this.SetMdop(), null, rate + 1000, rate);
        }
 
        /// <summary>Initializes a new instance of the ResponseLimitedConcurrencyTaskScheduler class with the specified MDOP function.</summary>
        /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism function provider for this scheduler.</param>
        public ResponseLimitedConcurrencyTaskScheduler(Func<int, int> funcMdop)
            : this(funcMdop, new ResponseLimitedConcurrencyTaskConfiguration())
        {
        }
 
        /// <summary> Define the MDOP based on the current state of the execution times. </summary>
        private void SetMdop()
        {
            if (this.timings != null && this.timings.Count > this.configuration.MinimumSampleCount)
            {
                lock (this.timingsSync)
                {
                    if (this.timings != null && this.timings.Count > this.configuration.MinimumSampleCount)
                    {
                        int[] latestTimes = new int[this.configuration.AverageSampleCount];
                        int count = this.timings.TryPopRange(latestTimes);
 
                        if (count > this.configuration.MinimumSampleCount)
                        {
                            this.timings.Clear();
                            int average = (int)latestTimes.Take(count).Average();
                            int mdop = this.funcMdop(average);
 
                            this.lastSampledAverage = average;
                            if (mdop > 0) { this.maxDegreeOfParallelism = mdop <= 0 ? this.configuration.DefaultDegreeOfParallelism : mdop; }
                        }
                    }
                }
            }
        }
 
        /// <summary>Attempts to execute the specified task recording execution time.</summary>
        /// <param name="task">The task to be executed.</param>
        /// <returns>Whether the task could be executed on the current thread.</returns>
        private bool TryExecuteTaskInternal(Task task)
        {
            Stopwatch sw = Stopwatch.StartNew();
 
            bool result = base.TryExecuteTask(task);
 
            sw.Stop();
            TimeSpan timing = sw.Elapsed;
 
            // Ensure queue is not growing too much before adding timing
            if (this.timings.Count < this.configuration.MaximumSampleCount) { this.timings.Push((int)timing.TotalMilliseconds); }
 
            return result;
        }
 
        /// <summary>Queues a task to the scheduler.</summary>
        /// <param name="task">The task to be queued.</param>
        protected override void QueueTask(Task task)
        {
            // Add the task to the list of tasks to be processed.  If there aren't enough
            // delegates currently queued or running to process tasks, schedule another.
            lock (this.tasks)
            {
                this.tasks.AddLast(task);
                if (this.delegatesQueuedOrRunning < this.maxDegreeOfParallelism)
                {
                    ++this.delegatesQueuedOrRunning;
                    this.NotifyThreadPoolOfPendingWork();
                }
            }
        }
 
        /// <summary> Informs the ThreadPool that there's work to be executed for this scheduler. </summary>
        private void NotifyThreadPoolOfPendingWork()
        {
            ThreadPool.UnsafeQueueUserWorkItem(_ =>
            {
                // Note that the current thread is now processing work items.
                // This is necessary to enable inlining of tasks into this thread.
                CurrentThreadIsProcessingItems = true;
                try
                {
                    // Process all available items in the queue.
                    while (true)
                    {
                        Task item;
                        lock (this.tasks)
                        {
                            // When there are no more items to be processed,
                            // or if running with too much concurrency,
                            // note that we're done processing, and get out.
                            if (this.tasks.Count == 0 || this.delegatesQueuedOrRunning > this.maxDegreeOfParallelism)
                            {
                                --this.delegatesQueuedOrRunning;
                                break;
                            }
 
                            // Get the next item from the queue
                            item = this.tasks.First.Value;
                            this.tasks.RemoveFirst();
                        }
 
                        // Execute the task we pulled out of the queue
                        this.TryExecuteTaskInternal(item);
                    }
                }
 
                // We're done processing items on the current thread
                finally
                {
                    CurrentThreadIsProcessingItems = false;
                }
            }, null);
        }
 
        /// <summary>Attempts to execute the specified task on the current thread.</summary>
        /// <param name="task">The task to be executed.</param>
        /// <param name="taskWasPreviouslyQueued">Indicator for whether the task was previously dequeued.</param>
        /// <returns>Whether the task could be executed on the current thread.</returns>
        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        {
            // If this thread isn't already processing a task, we don't support inlining
            if (!CurrentThreadIsProcessingItems) { return false; }
 
            // If the task was previously queued, remove it from the queue
            if (taskWasPreviouslyQueued) this.TryDequeue(task);
 
            // Try to run the task.
            return this.TryExecuteTaskInternal(task);
        }
 
        /// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
        /// <param name="task">The task to be removed.</param>
        /// <returns>Whether the task could be found and removed.</returns>
        protected override bool TryDequeue(Task task)
        {
            lock (this.tasks)
            {
                return this.tasks.Remove(task);
            }
        }
 
        /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
        public override int MaximumConcurrencyLevel
        {
            get
            {
                return this.maxDegreeOfParallelism;
            }
        }
 
        /// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
        /// <returns>An enumerable of the tasks currently scheduled.</returns>
        protected override IEnumerable<Task> GetScheduledTasks()
        {
            bool lockTaken = false;
            try
            {
                Monitor.TryEnter(this.tasks, ref lockTaken);
 
                if (lockTaken)
                {
                    return this.tasks.ToArray();
                }
                else
                {
                    throw new NotSupportedException();
                }
            }
            finally
            {
                if (lockTaken) { Monitor.Exit(this.tasks); }
            }
        }
 
        /// <summary>Gets the current Execution State of the scheduler.</summary>
        public ExecutionState SchedulerExecutionState
        {
            get
            {
                ExecutionState state = new ExecutionState();
                state.LastSampledAverage = this.lastSampledAverage;
                state.MaxDegreeOfParallelism = this.maxDegreeOfParallelism;
                state.CurrentConcurrency = this.delegatesQueuedOrRunning;
                state.AwaitingTasks = this.tasks.Count;
 
                return state;
            }
        }
 
        /// <summary>Gets the current Configuration of the scheduler.</summary>
        public ResponseLimitedConcurrencyTaskConfiguration SchedulerTaskConfiguration
        {
            get
            {
                return this.configuration;
            }
        }
 
        /// <summary>Gets a value indicating whether the scheduler is processing.</summary>
        public bool IsProcessing
        {
            get
            {
                return this.tasks.Count > 0 || this.delegatesQueuedOrRunning > 0;
            }
        }
 
        /// <summary>Method to control the disposing of resources. </summary>
        /// <param name="disposing">Indicator for whether Dispose has been called directly.</param>
        private void Dispose(bool disposing)
        {
            if (!this.disposed)
            {
                if (disposing)
                {
                    if (this.refreshTimer != null)
                    {
                        this.refreshTimer.Dispose();
                    }
                }
 
                // Indicate that the instance has been disposed.
                this.refreshTimer = null;
                this.disposed = true;
            }
        }
 
        /// <summary>IDisposable Dispose method.</summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }
    }
}

To use the task scheduler one just has to create an instance, with the required MDOP function, and then as before create a Factory with the newly created scheduler:

var config = new List<ResponseLimitedConcurrencyStepDefinition>();

config.Add(new ResponseLimitedConcurrencyStepDefinition(5, 500));

config.Add(new ResponseLimitedConcurrencyStepDefinition(3, 5000));

config.Add(new ResponseLimitedConcurrencyStepDefinition(2, Int32.MaxValue));

Func<int, int> mdopFunc = ResponseLimitedConcurrencyFunctionFactory.StepDecrement(config);

ResponseLimitedConcurrencyTaskScheduler rlcts = new ResponseLimitedConcurrencyTaskScheduler(mdopFunc);

TaskFactory factory = new TaskFactory(rlcts);

factory.StartNew(() =>

    {

        for (int i = 0; i < 10; i++)

        {

            Console.WriteLine("{0} on thread {1}", i, Thread.CurrentThread.ManagedThreadId);

        }

    }

);

The only difference in this code is the construction of the MDOP function. To assist in creating this function the code comes with a few starters for 10. Firstly a step up and step down function, and secondly a linearly increasing and decreasing function. The step up and down functions takes a series of steps definitions outlining the step changes in MDOP. The linear increasing and decreasing functions take an upper and lower limit with a rate of change.

There is one final item needed to meet the requirements that initially drove the custom task scheduler solution. How do you use the scheduler to process a queue of actions?

ResponseLimitedConcurrencyActionLoader Class

The downloadable code also contains a class that can be used to ensure the task scheduler instance always has an active processing queue of a specified Action:

namespace MSDN.Schedulers
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
 
    /// <summary>A class to load a ResponseLimitedConcurrencyTaskScheduler with a single Action.</summary>
    public class ResponseLimitedConcurrencyActionLoader
    {
        /// <summary>Indicator for whether to continue loading.</summary>
        private volatile bool continueProcessing = false;
 
        /// <summary>The ResponseLimitedConcurrencyTaskScheduler reference.</summary>
        private ResponseLimitedConcurrencyTaskScheduler scheduler;
 
        /// <summary>The factory to be used for loading the action.</summary>
        private TaskFactory factory;
 
        /// <summary>The Task used to load the data on a background thread.</summary>
        private Task loaderTask;
 
        /// <summary>The interval (in milliseconds) for which to review the outstanding tasks.</summary>
        private int baseInterval;
 
        /// <summary>Initializes a new instance of the ResponseLimitedConcurrencyActionLoader class.</summary>
        /// <param name="scheduler">The scheduler to load.</param>
        /// <param name="interval">The interval (in milliseconds) for which to perform a load check.</param>
        public ResponseLimitedConcurrencyActionLoader(ResponseLimitedConcurrencyTaskScheduler scheduler, int interval)
        {
            this.scheduler = scheduler;
            this.factory = new TaskFactory(scheduler);
            this.baseInterval = interval;
        }
 
        /// <summary>Initializes a new instance of the ResponseLimitedConcurrencyActionLoader class.</summary>
        /// <param name="scheduler">The scheduler to load.</param>
        /// <param name="interval">The interval (in milliseconds) for which to perform a load check.</param>
        public ResponseLimitedConcurrencyActionLoader(ResponseLimitedConcurrencyTaskScheduler scheduler)
            : this(scheduler, 1000)
        {
        }
 
        /// <summary>Indicators whether the scheduler or loader is currently busy.</summary>
        public bool IsProcessing()
        {
            return this.continueProcessing || this.scheduler.IsProcessing;
        }
 
        /// <summary>Performs a load of the specified action until stop is called.</summary>
        /// <param name="action">The action to execute.</param>
        public void Start(Action action)
        {
            // Ensure loading is currently not active
            if (this.IsProcessing()) { throw new InvalidOperationException("Loader is currently active."); }
 
            // Note now running
            this.continueProcessing = true;
 
            // Define base configuration values
            var configuration = this.scheduler.SchedulerTaskConfiguration;
            int minSampleRate = Math.Min(configuration.MinimumSampleCount * 3, configuration.AverageSampleCount * 2);
            int interval = Math.Min(configuration.SamplingRateMilliseconds, this.baseInterval);
 
            // Load actions into scheduler until processing needs to stop
            this.loaderTask = Task.Factory.StartNew(() =>
                {
                    while (this.continueProcessing)
                    {
                        // Calculate tasks that should be pending
                        int actualNeeded;
                        if (this.scheduler.SchedulerExecutionState.LastSampledAverage == 0)
                        {
                            double rate = ((double)this.scheduler.SchedulerExecutionState.MaxDegreeOfParallelism * 1000.0) / (double)this.scheduler.SchedulerExecutionState.LastSampledAverage;
                            int minNeeded = Math.Max((int)((rate * (double)interval * 2.0) / 1000.0), minSampleRate);
                            actualNeeded = minNeeded - this.scheduler.SchedulerExecutionState.AwaitingTasks;
                        }
                        else
                        {
                            actualNeeded = minSampleRate;
                        }
 
                        // Load any needed tasks
                        if (actualNeeded > 0)
                        {
                            for (int idx = 0; idx < actualNeeded; idx++)
                            {
                                factory.StartNew(action);
                            }
                        }
 
                        // Wait for inteval before looping back around
                        Thread.Sleep(interval);
                    }
                });
        }
 
        /// <summary>Tell the loader to stop processing.</summary>
        public void Stop()
        {
            this.continueProcessing = false;
            if (this.loaderTask != null) { this.loaderTask.Wait(); }
        }
 
        /// <summary>Tell the loader to stop processing and block until all actions complete.</summary>
        public void StopAndWait()
        {
            this.Stop();
            while (this.scheduler.IsProcessing) { Thread.Sleep(250); }
        }
    }
}

If needed this class can be used to construct a Windows Service that processes items from a queue, and where the MDOP of the scheduler varies based on the workload. It is for this reason the code contains Start() and Stop() function calls. To use the Action Loader one just has to provide the necessary Action and Scheduler to the class instance:

Func<int, int> funcMdop = ResponseLimitedConcurrencyFunctionFactory.LinearDecreasing(5, 1, 2);

ResponseLimitedConcurrencyTaskScheduler rlcts = new ResponseLimitedConcurrencyTaskScheduler(funcMdop);

ResponseLimitedConcurrencyActionLoader loader = new ResponseLimitedConcurrencyActionLoader(rlcts);

Action action = () =>

        {

            Thread.Sleep(10);

            Console.WriteLine("Finish Thread={0}", Thread.CurrentThread.ManagedThreadId);

        };

loader.Start(action);

Thread.Sleep(5000);

loader.StopAndWait();

So going back to my original requirements, it became easy to create a Windows Service where the Action is to read from a processing queue and call a web service; where the MDOP is reduced when the service response times increase.

Summary

There are a variety of possible approaches one could take for Action processing, where the number of processing threads is varied based on task execution times. I chose to explore the custom task Scheduler approach as it offered the greatest amount of flexibility and reuse for a variety of other possible scenarios.

As a final reminder the code download can be located here.

Framework for .Net Hadoop MapReduce Job Submission TextOutput Type

2012-07-10T21:10:10Z

Some recent changes made to the “Generics based Framework for .Net Hadoop MapReduce Job Submission” code were to support Json and Binary Serialization from Mapper, in and out of Combiners, and out from the Reducer. However, this precluded one from controlling the format of the Text output. Say one wanted to create a tab delimited string from the Reducer. This could only be done using Json Serialization. To better support allowing one to construct the final text output I have created a new TextOutput type.

This TextOutput type is simple in structure. However, when this type is encountered during the serialization process, both Json and Binary serialization are bypassed and the text is written out in its raw format; including tabs and other characters usually escaped by the Json serializer.

As an example here is a modified version of one of the C# Reducer samples that supports both Json and Text output:

namespace MSDN.Hadoop.MapReduceCSharp
 {
     [DataContract]
     public class MobilePhoneRange
     {
         [DataMember] public TimeSpan MinTime { get; set; }
         [DataMember] public TimeSpan MaxTime { get; set; }
 
         public MobilePhoneRange(TimeSpan minTime, TimeSpan maxTime)
         {
             this.MinTime = minTime;
             this.MaxTime = maxTime;
         }
 
         public TextOutput ToText(string format)
         {
             return new TextOutput(String.Format(@"({0}, {1})", this.MinTime.ToString(format), this.MaxTime.ToString(format)));
         }
     }
 
     public class MobilePhoneRangeReducer : ReducerBase<TimeSpan, MobilePhoneRange>
     {
         public override IEnumerable<Tuple<string, MobilePhoneRange>> Reduce(string key, IEnumerable<TimeSpan> value)
         {
             var baseRange = new MobilePhoneRange(TimeSpan.MaxValue, TimeSpan.MinValue);
             var rangeValue = value.Aggregate(baseRange, (accSpan, timespan) =>
                 new MobilePhoneRange((timespan < accSpan.MinTime) ? timespan : accSpan.MinTime, (timespan > accSpan.MaxTime ) ? timespan : accSpan.MaxTime));
 
             yield return new Tuple<string, MobilePhoneRange>(key, rangeValue);
         }
     }
 
     public class MobilePhoneRangeTextReducer : ReducerBase<TimeSpan, TextOutput>
     {
         public override IEnumerable<Tuple<string, TextOutput>> Reduce(string key, IEnumerable<TimeSpan> value)
         {
             var baseRange = new MobilePhoneRange(TimeSpan.MaxValue, TimeSpan.MinValue);
             var rangeValue = value.Aggregate(baseRange, (accSpan, timespan) =>
                 new MobilePhoneRange((timespan < accSpan.MinTime) ? timespan : accSpan.MinTime, (timespan > accSpan.MaxTime) ? timespan : accSpan.MaxTime));
 
             yield return new Tuple<string, TextOutput>(key, rangeValue.ToText("G"));
         }
     }
 }

For the sample Reducer above the Json serialization output would be:

Android {"MaxTime":"PT23H59M54S","MinTime":"PT6S"}
RIM OS {"MaxTime":"PT23H59M58S","MinTime":"PT1M7S"}
Unknown {"MaxTime":"PT23H52M36S","MinTime":"PT36S"}
Windows Phone {"MaxTime":"PT23H55M17S","MinTime":"PT32S"}
iPhone OS {"MaxTime":"PT23H59M50S","MinTime":"PT1S"}

The corresponding Text Output would be:

Android (0:00:00:06.0000000, 0:23:59:54.0000000)
RIM OS (0:00:01:07.0000000, 0:23:59:58.0000000)
Unknown (0:00:00:36.0000000, 0:23:52:36.0000000)
Windows Phone (0:00:00:32.0000000, 0:23:55:17.0000000)
iPhone OS (0:00:00:01.0000000, 0:23:59:50.0000000)

As mentioned the actual definition of the TextOutput type is simple and is just a wrapper over a string, although depending on needs this may change:

type TextOutput(value:string) = 
 
     // Internal text value
     let mutable text = value
 
     /// String value of the TextOutput class
     member this.Text
         with get () = text
         and set (value) = text <- value
 
     /// Byte array value of the TextOutput class
     member this.Bytes
         with get() = Encoding.UTF8.GetBytes(text)
         and set (value:byte array) = text <- Encoding.UTF8.GetString(value)
 
     new(value:byte array) = 
         TextOutput(Encoding.UTF8.GetString(value))
 
     new(value:TextOutput) =
         TextOutput(value.Text)
 
     new() = 
         TextOutput(String.Empty)
 
     /// Clear Text 
     member this.Clear() = 
         text <- String.Empty
 
     /// Append Text
     member this.Append(value:string) = 
         text <- text + value
         text
 
     /// ToString override
     override this.ToString() =
         text
 
     ///Equals override
     override this.Equals(value:obj) =
         if not (value.GetType() = typeof<TextOutput>) then
             false
         else
             let objText = (value :?> TextOutput)
             text.Equals(objText.Text)
 
     /// GetHasCode override
     override this.GetHashCode() =
         text.GetHashCode()

One of the main rationales for adding the TextOutput support is so that data output by the framework can be easily used by Hive CREATE TABLE statements.

Hope you find this change useful.

C# MapReduce Based Co-occurrence Item Based Recommender

2012-07-09T19:17:20Z

As promised, to conclude the Co-occurrence Approach to an Item Based Recommender posts I wanted to port the MapReduce code to C#; just for kicks and to prove the code is also easy to write in C#. For an explanation of the MapReduce post review the previous article:

http://blogs.msdn.com/b/carlnol/archive/2012/07/07/mapreduce-based-co-occurrence-approach-to-an-item-based-recommender.aspx

The latest version of the code now includes the MSDN.Recommender.MapReduceCs project, that is a full C# implementation of the corresponding F# project; MSDN.Recommender.MapReduce. These code variants are intentionally similar in structure.

The code for the Order Mapper is as follows:

Order Mapper
 public class OrderVectorMapper : MapperBaseText<Types.ProductQuantityList>
{
    // Configuration values
    static double qtyMaximum = 5.0;                      // Maximum rating contribution for an item
    static double recentFactor = 2.0;                    // Quantity increase factor for recent items
    
    static DateTime baseDate = DateTime.Today.AddMonths(-3);    // Date for a recent item
 
    Types.ProductQuantityList products = new Types.ProductQuantityList();
    int? currentOrder = null;
 
    // Converts an order Id to a string key
    private static string getOrderKey(int orderId)
    {
        return String.Format("{0:D}", orderId);
    }
 
    // Adds a quantity factor based on recent orders
    private static double orderFactor(Types.OrderDetail order)
    {
        if (DateTime.Compare(order.OrderDate, OrderVectorMapper.baseDate) > 0)
        {
            return OrderVectorMapper.recentFactor;
        }
        else
        {
            return 1.0;
        }
    }
 
    /// Map the data from input name/value to output name/value
    public override IEnumerable<Tuple<string, Types.ProductQuantityList>> Map(string value)
    {
        // Ensure order can be derived
        Types.OrderDetail order = null;
        try
        {
            order = Types.Helpers.ParseInputData(value);
        }
        catch (ArgumentException)
        {
            order = null;
        }
 
        // Process the order
        if (order != null)
        {
            double quantity = Math.Min(((double)order.OrderQty), OrderVectorMapper.qtyMaximum) * OrderVectorMapper.orderFactor(order);
            Types.ProductQuantity product = new Types.ProductQuantity(order.ProductId, quantity);
 
            if (this.currentOrder.HasValue && (order.OrderId != this.currentOrder.Value))
            {
                yield return Tuple.Create(OrderVectorMapper.getOrderKey(currentOrder.Value), this.products);
                this.products.Clear();
            }
            this.currentOrder = order.OrderId;
            this.products.Add(product);
        }
        else
        {
            Context.IncrementCounter("ORDERS", "Skipped Lines");
        }
    }
 
    /// Output remaining Map items
    public override IEnumerable<Tuple<string, Types.ProductQuantityList>> Cleanup()
    {
        if (this.currentOrder.HasValue)
        {
            yield return Tuple.Create(OrderVectorMapper.getOrderKey(currentOrder.Value), this.products);
        }
    }
}
 

And that of the Order Reducer is:

Order Reducer
 class OrderVectorReducer : ReducerBase<Types.ProductQuantityList, Types.ProductQuantityList>
{
    /// Reduce the order data into a product list
    public override IEnumerable<Tuple<string, Types.ProductQuantityList>> Reduce(string key, IEnumerable<Types.ProductQuantityList> values)
    {
        var products = new Types.ProductQuantityList();
 
        foreach (var prodList in values)
        {
            products.AddRange(prodList);
        }
 
        yield return Tuple.Create(key, products);
    }
}
 

Your friend when writing the C# code is the yield return command for emitting data, and the Tuple.Create() command for creating the key/value pairs. The two commands enable one to return a key/value sequence; namely a .Net IEnumerable.

For completeness the code for the Product Mapper is:

Product Mapper
 class ProductVectorMapper : MapperBaseText<Types.ProductQuantity>
{
    static int maxSize = 1024 * 1024;
    Dictionary<Tuple<int, int>, double> prodPairs = new Dictionary<Tuple<int, int>, double>(ProductVectorMapper.maxSize);
 
    // Converts an order Id to a string key
    static private string getProductKey(int productId)
    {
        return String.Format("{0:D}", productId);
    }
 
    // Parses an input line of format List<ProductQuantity>
    static private Types.ProductQuantityList deserialize(string input)
    {
        string[] keyValue = input.Split('\t');
        return Types.Helpers.ParseProductQuantityList(keyValue[1].Trim());
    }
 
    // calculates the pairs for an order
    static private IEnumerable<Tuple<T, T>> pairs<T>(List<T> items)
    {
        int count = items.Count;
        if (count == 2)
        {
            yield return Tuple.Create(items[0], items[1]);
        }
        else if (count > 2)
        {
            for (int idxOut = 0; idxOut <= (count - 2); idxOut++)
            {
                for (int idxIn = (idxOut + 1); idxIn <= (count - 1); idxIn++)
                {
                    yield return Tuple.Create(items[idxOut], items[idxIn]);
                }
            }
        }
    }
 
    // Adds to the table
    private void addRow(Tuple<int, int> idx, double qty)
    {
        if (this.prodPairs.ContainsKey(idx))
        {
            this.prodPairs[idx] = this.prodPairs[idx] + qty;
        }
        else
        {
            this.prodPairs[idx] = qty;
        }
    }
 
    // Defines a sequence of the current pairs
    private IEnumerable<Tuple<string, Types.ProductQuantity>> currents()
    {
        foreach (var item in prodPairs)
        {
            int product1 = item.Key.Item1;
            int product2 = item.Key.Item2;
            yield return Tuple.Create(ProductVectorMapper.getProductKey(product1), new Types.ProductQuantity(product2, item.Value));                
        }
        prodPairs.Clear();
    }
 
    /// Map the data from input name/value to output name/value
    public override IEnumerable<Tuple<string, Types.ProductQuantity>> Map(string value)
    {
        var products = ProductVectorMapper.pairs<Types.ProductQuantity>(ProductVectorMapper.deserialize(value));
        foreach (var product in products)
        {
            Types.ProductQuantity product1 = product.Item1;
            Types.ProductQuantity product2 = product.Item2;
            double qty = Math.Max(product1.Quantity, product2.Quantity);
            this.addRow(Tuple.Create(product1.ProductId, product2.ProductId), qty);
            this.addRow(Tuple.Create(product2.ProductId, product1.ProductId), qty);
        }
 
        if (prodPairs.Count > ProductVectorMapper.maxSize)
        {
            return currents();
        }
        else
        {
            return Enumerable.Empty<Tuple<string, Types.ProductQuantity>>();
        }
    }
 
    /// Output remaining Map items
    public override IEnumerable<Tuple<string, Types.ProductQuantity>> Cleanup()
    {
        return currents();
    }
}
 

That of the Product Combiner is:

Product Combiner
 class ProductVectorCombiner : CombinerBase<Types.ProductQuantity>
{
    /// Combine the data from input name/value to output name/value
    public override IEnumerable<Tuple<string, Types.ProductQuantity>> Combine(string key, IEnumerable<Types.ProductQuantity> values)
    {
        int maxSize = 10000;               // Expected number of product correlations
        var products = new Dictionary<int, double>(maxSize);
 
        // Process the combiner input adding to the table
        foreach (var product in values)
        {
            if (products.ContainsKey(product.ProductId))
            {
                products[product.ProductId] = products[product.ProductId] + product.Quantity;
            }
            else
            {
                products[product.ProductId] = product.Quantity;
            }
        }
 
        // Return the combiner input
        return products.Select(item => Tuple.Create(key, new Types.ProductQuantity(item.Key, item.Value)));
    }
}
 

And finally, that of the Product Reducer is:

Product Reducer
 class ProductVectorReducer : ReducerBase<Types.ProductQuantity, Types.ProductRecommendations>
{
    /// Reduce the data from input name/value to output name/value
    public override IEnumerable<Tuple<string, Types.ProductRecommendations>> Reduce(string key, IEnumerable<Types.ProductQuantity> values)
    {
        // Configuration values
        double entryThreshold = 20.0;                     // Minimum correlations for matrix inclusion
        int matrixSize = 10000;                           // Expected Correlations for Hash Table init
 
        int minItem = Int32.MaxValue;
        int maxItem = 0;
        var rcTable = new Dictionary<int, double>(matrixSize);
 
        // Process the combiner input adding to the table
        foreach (var product in values)
        {
            int idx = product.ProductId;
            minItem = Math.Min(idx, minItem);
            maxItem = Math.Max(idx, maxItem);
 
            if (rcTable.ContainsKey(product.ProductId))
            {
                rcTable[product.ProductId] = rcTable[product.ProductId] + product.Quantity;
            }
            else
            {
                rcTable[product.ProductId] = product.Quantity;
            }
        }
 
        int offset = minItem;
        int size = maxItem + 1 - minItem;
 
        // Build the sparse vector
        var vector = new SparseVector(size);
        foreach (var item in rcTable)
        {
            if (item.Value > entryThreshold)
            {
                vector[item.Key - offset] = item.Value;
            }
        }
 
        var recommendations = new Types.ProductRecommendations(Int32.Parse(key), offset, vector);
        Context.IncrementCounter("PRODUCTS", "Recommendations Written");
 
        yield return Tuple.Create(key, recommendations);
    }
}
 

A lot of the processing is based upon processing IEnumerable sequences. For this reason foreach and Linq expressions are used a lot. I could have defined a ForEach IEnumerable extension, to use Linq exclusively, but have preferred the C# syntax.

One difference in the final Reducer is how the Sparse Matrix is defined. In the F# code an F# extension allows one to build a SparseMatrix type by providing a sequence of index/value tuples. In the C# code one has to define an empty SparseMatrix and then interactively set the element values.

Note: There may be some confusion in the usage of the float type for .Net programmers, when switching between C# and F#, due to F#’s naming for floating-point number types. F# float type corresponds to System.Double in .NET (which is called double in C#), and F# float32 type corresponds to System.Single in .NET (which is called float in C#). For this reason I have switched to using the type double in F#, as it is another type alias for the System.Double type.

Of course the recommendations produced by the C# code are the same as the F# code, and clearly demonstrate that the Framework cleanly support writing MapReduce jobs in C#. I personally find the F# code more succinct and easier to read, but this is my personal preference.