Extra examples from Musketeer work

Examples/DifferentialDataflow/Netflix.cs

@@ -0,0 +1,222 @@
+using Microsoft.Research.Naiad;
+using Microsoft.Research.Naiad.Dataflow;
+using Microsoft.Research.Naiad.Frameworks.DifferentialDataflow;
+using Microsoft.Research.Naiad.Input;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Text;
+
+namespace Microsoft.Research.Naiad.Examples.DifferentialDataflow {
+
+  public class Netflix : Example {
+
+    int movieSelect = 1970;
+
+    public struct IntTriple : IEquatable<IntTriple> {
+      public int first;
+      public int second;
+      public int third;
+
+      public bool Equals(IntTriple that) {
+        return first == that.first && second == that.second && third == that.third;
+      }
+
+      public int CompareTo(IntTriple that) {
+        if (this.first != that.first)
+          return this.first - that.first;
+
+        if (this.second != that.second)
+          return this.second - that.second;
+
+        return this.third - that.third;
+      }
+
+      // Embarassing hashcodes
+      public override int GetHashCode() {
+        return first + 1234347 * second + 4311 * third;
+      }
+
+      public override string ToString() {
+        return String.Format("{0} {1} {2}", first, second, third);
+      }
+
+      public IntTriple(int x, int y, int z)
+      {
+        first = x; second = y; third = z;
+      }
+    }
+
+    public struct Movie : IEquatable<Movie> {
+      public int id;
+      public int year;
+      public string title;
+
+      public bool Equals(Movie that) {
+        return id == that.id && year == that.year && title.Equals(that.title);
+      }
+
+      public int CompareTo(Movie that) {
+        if (id != that.id)
+          return id - that.id;
+        if (year != that.year)
+          return year - that.year;
+        return title.CompareTo(that.title);
+      }
+
+      public override int GetHashCode() {
+        return 31 * id + 1234347 * year + 4311 * title.GetHashCode();
+      }
+
+      public override string ToString() {
+        return String.Format("{0} {1} {2}", id, year, title);
+      }
+
+      public Movie(int x, int y, string z) {
+        id = x; year = y; title = z;
+      }
+
+    }
+
+    public struct Rating : IEquatable<Rating> {
+      public int movId;
+      public int userId;
+      public int rating;
+      public string date;
+
+      public bool Equals(Rating that) {
+        return movId == that.movId && userId == that.userId &&
+          rating == that.rating && date.Equals(that.date);
+      }
+
+      public int CompareTo(Rating that) {
+        if (movId != that.movId)
+          return movId - that.movId;
+        if (userId != that.userId)
+          return userId - that.userId;
+        if (rating != that.rating)
+          return rating - that.rating;
+        return date.CompareTo(that.date);
+      }
+
+      public override int GetHashCode() {
+        return 31 * movId + 1234347 * userId + 4311 * rating + 12315 *
+          date.GetHashCode();
+      }
+
+      public override string ToString() {
+        return String.Format("{0} {1} {2} {3}", movId, userId, rating,
+                             date);
+      }
+
+      public Rating(int x, int y, int z, string w) {
+        movId = x;
+        userId = y;
+        rating = z;
+        date = w;
+      }
+
+    }
+
+    public IEnumerable<IntTriple> SumRatings(IntPair movIds,
+                                             IEnumerable<int> ratings) {
+      int allRatings = 0;
+      foreach (var rating in ratings)
+        allRatings += rating;
+      yield return new IntTriple(movIds.s, movIds.t, allRatings);
+    }
+
+    public IEnumerable<IntPair> MaxRating(int id, IEnumerable<int> ratings) {
+      int maxRating = 0;
+      foreach (var rating in ratings)
+        maxRating = Math.Max(maxRating, rating);
+      yield return new IntPair(id, maxRating);
+    }
+
+    public void Execute(string[] args) {
+      using (var computation = NewComputation.FromArgs(ref args)) {
+        var movies = computation.NewInputCollection<Movie>();
+        var ratings = computation.NewInputCollection<Rating>();
+
+        var ratingSel = movies.Where((Movie x) => x.year < 1970)
+          .Join(ratings,
+                (Movie movie) => movie.id,
+                (Rating rating) => rating.movId,
+                (Movie movie, Rating rating) =>
+                new IntTriple(rating.movId, rating.userId, rating.rating));
+
+        var ratingJoin = ratingSel.Join(ratingSel,
+                                        (IntTriple r1) => r1.second,
+                                        (IntTriple r2) => r2.second,
+                                        (IntTriple r1, IntTriple r2) =>
+                                        new IntTriple(r1.first, // movId
+                                                      r1.third * r2.third, // rating * rating
+                                                      r2.first)); // movId
+
+        var intResult = ratingJoin.GroupBy(row => new IntPair(row.first, row.third),
+                                           row => row.second, // Select rating
+                                           (movIds, grouped_ratings) =>
+                                           SumRatings(movIds, grouped_ratings));
+
+        var matmultfinalPrj = intResult.Join(ratingSel,
+                       (IntTriple intRes) => intRes.first, // movId
+                       (IntTriple ratSel) => ratSel.first, // movId
+                       (IntTriple intRes, IntTriple ratSel) =>
+                       new IntTriple(intRes.first,
+                                     intRes.third * ratSel.third,
+                                     ratSel.second));
+
+        var predicted = matmultfinalPrj.GroupBy(row => new IntPair(row.first, row.third),
+                                                row => row.second, // Select rating
+                                                (movIds, grouped_ratings) =>
+                                                SumRatings(movIds, grouped_ratings));
+
+        var prediction = predicted.GroupBy(row => row.second,
+                                           row => row.third,
+                                           (id, movId) => MaxRating(id, movId))
+          .Subscribe(l => { foreach (var element in l) Console.WriteLine(element); });
+
+        computation.Activate();
+        var inputFile = args[1];
+        if (computation.Configuration.ProcessID == 0) {
+          StreamReader reader = File.OpenText(inputFile);
+
+          for (var i = 0; !reader.EndOfStream; i++) {
+            var elements = reader.ReadLine().Split(' ');
+            if (elements.Length == 3) {
+              movies.OnNext(new Movie(Convert.ToInt32(elements[0]),
+                                      Convert.ToInt32(elements[1]),
+                                      elements[2]));
+              ratings.OnNext();
+              computation.Sync(i);
+            }
+            if (elements.Length == 4) {
+              movies.OnNext();
+              ratings.OnNext(new Rating(Convert.ToInt32(elements[0]),
+                                        Convert.ToInt32(elements[1]),
+                                        Convert.ToInt32(elements[2]),
+                                        elements[3]));
+              computation.Sync(i);
+            }
+          }
+        }
+
+        movies.OnCompleted();
+        ratings.OnCompleted();
+
+        computation.Join();
+      }
+    }
+
+    public string Usage { get { return ""; } }
+
+    public string Help {
+      get {
+        return "Netflix <input file>";
+      }
+    }
+  }
+
+}

Examples/Examples.csproj

@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="utf-8"?>
+<?xml version="1.0" encoding="utf-8"?>
 <Project ToolsVersion="4.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
   <Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
   <PropertyGroup>
@@ -74,15 +74,26 @@
     </Compile>
     <Compile Include="DifferentialDataflow\ConnectedComponents.cs" />
     <Compile Include="DifferentialDataflow\GraphColoring.cs" />
+    <Compile Include="DifferentialDataflow\Netflix.cs" />
     <Compile Include="DifferentialDataflow\SearchIndex.cs" />
     <Compile Include="DifferentialDataflow\StronglyConnectedComponents.cs" />
     <Compile Include="DifferentialDataflow\WordCount.cs" />
     <Compile Include="GraphLINQ\PageRank.cs" />
     <Compile Include="GraphLINQ\Reachability.cs" />
     <Compile Include="Naiad\ConnectedComponents.cs" />
     <Compile Include="Naiad\KeyValueLookup.cs" />
+    <Compile Include="Naiad\Join.cs" />
+    <Compile Include="Naiad\JoinLJ.cs" />
+    <Compile Include="Naiad\KMeans.cs" />
     <Compile Include="Naiad\Latency.cs" />
+    <Compile Include="Naiad\Netflix.cs" />
+    <Compile Include="Naiad\PageRank.cs" />
+    <Compile Include="Naiad\Project.cs" />
+    <Compile Include="Naiad\ReadTest.cs" />
+    <Compile Include="Naiad\Shopper.cs" />
+    <Compile Include="Naiad\SSSP.cs" />
     <Compile Include="Naiad\Throughput.cs" />
+    <Compile Include="Naiad\TPCH.cs" />
     <Compile Include="Naiad\WordCount.cs" />
     <Compile Include="Program.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
@@ -116,4 +127,4 @@
   <Target Name="AfterBuild">
   </Target>
   -->
-</Project>
+</Project>

Examples/Naiad/Join.cs

@@ -0,0 +1,131 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.IO;
+using System.Text;
+
+using Microsoft.Research.Naiad;
+using Microsoft.Research.Naiad.Dataflow;
+using Microsoft.Research.Naiad.Frameworks.DifferentialDataflow;
+using Microsoft.Research.Naiad.Frameworks.Lindi;
+using Microsoft.Research.Naiad.Input;
+
+namespace Microsoft.Research.Naiad.Examples.Join
+{
+
+  public class Join : Example
+  {
+    public struct IntPair : IEquatable<IntPair> {
+      public int s;
+      public int t;
+
+      public bool Equals(IntPair that) {
+        return s == that.s && t == that.t;
+      }
+
+      public int CompareTo(IntPair that) {
+        if (this.s != that.s)
+          return this.s - that.s;
+
+        return this.t - that.t;
+      }
+
+      public override int GetHashCode() {
+        return 47 * s + 36425232 * t;
+      }
+
+      public override string ToString() {
+        return String.Format("{0} {1}", s, t);
+      }
+
+      public IntPair(int ss, int tt) {
+        s = ss;
+        t = tt;
+      }
+    }
+
+    public struct IntTriple : IEquatable<IntTriple> {
+      public int first;
+      public int second;
+      public int third;
+
+      public bool Equals(IntTriple that) {
+        return first == that.first && second == that.second && third == that.third;
+      }
+
+      public int CompareTo(IntTriple that) {
+        if (this.first != that.first)
+          return this.first - that.first;
+
+        if (this.second != that.second)
+          return this.second - that.second;
+
+        return this.third - that.third;
+      }
+
+      // Embarassing hashcodes
+      public override int GetHashCode() {
+        return first + 1234347 * second + 4311 * third;
+      }
+
+      public override string ToString() {
+        return String.Format("{0} {1} {2}", first, second, third);
+      }
+
+      public IntTriple(int x, int y, int z) {
+        first = x; second = y; third = z;
+      }
+    }
+
+    public IEnumerable<IntPair> read_rel(string filename) {
+      StreamReader reader = File.OpenText(filename);
+      for (; !reader.EndOfStream; ) {
+        var elements = reader.ReadLine().Split(' ');
+        yield return new IntPair(Convert.ToInt32(elements[0]),
+                                 Convert.ToInt32(elements[1]));
+      }
+    }
+
+    public string Usage {get {return "<local path prefix>";} }
+
+    public void Execute(string[] args) {
+      using (var computation = NewComputation.FromArgs(ref args)) {
+        var left_input = new BatchedDataSource<string>();
+        var right_input = new BatchedDataSource<string>();
+        var left = computation.NewInput(left_input).SelectMany(filename => read_rel(filename));
+        var right = computation.NewInput(right_input).SelectMany(filename => read_rel(filename));
+
+        var output = left.Join(right,
+               left_row => left_row.t,
+               right_row => right_row.t,
+               (left_row, right_row) => new IntTriple(left_row.s, left_row.t, right_row.s));
+
+        int minThreadId = computation.Configuration.ProcessID *
+          computation.Configuration.WorkerCount;
+        StreamWriter[] file_output =
+          new StreamWriter[computation.Configuration.WorkerCount];
+        for (int i = 0; i < computation.Configuration.WorkerCount; ++i) {
+          int j = minThreadId + i;
+          file_output[i] = new StreamWriter(args[1] + "/join" + j + ".out");
+        }
+        output.Subscribe((i, l) => { foreach (var element in l) file_output[i - minThreadId].WriteLine(element); });
+        computation.Activate();
+        left_input.OnCompleted(args[1] + "/join_left" + computation.Configuration.ProcessID + ".in");
+        right_input.OnCompleted(args[1] + "/join_right" + computation.Configuration.ProcessID + ".in");
+        computation.Join();
+        for (int i = 0; i < computation.Configuration.WorkerCount; ++i) {
+          file_output[i].Close();
+        }
+      }
+    }
+
+    public string Help {
+      get {
+        return "Join <local path prefix>";
+      }
+    }
+  }
+
+}