Merge branch 'master' of github.com:lrsjohnson/scheme-mapreduce

Author: lrsjohnson

data-loader.scm

@@ -1,6 +1,5 @@
 ;; TODO: Make generic? Only works on list right now
 (define (mrs:feed-data data-set ds-elt-sequence)
-  (pp 'feed-data)
   (for-each (lambda (ds-elt)
 	      (ds-add-elt data-set ds-elt))
               ds-elt-sequence))
@@ -16,11 +15,16 @@
 	       (lp (+ i 1)
 		   (cdr value-list)))))))
 
+(define (mrs:feed-key-value-list data-set key-value-list)
+  (mrs:feed-data
+   data-set
+   (map (lambda (kv)
+          (create-ds-elt (car kv) (cadr kv))) key-value-list)))
+
 (define (mrs:run-computation thunk)
   (with-time-sharing-conspiracy
    (lambda ()
      (thunk)
-     (pp 'run-c)
      (flush-input-data-sets)
      (conspire:null-job))))
 

examples/demo.scm

@@ -0,0 +1,30 @@
+;;; Flow computation demo
+(define (demo)
+  (define 2d-vectors (create-data-set))
+  (define 3d-vectors (create-data-set))
+  (define magnitudes (create-data-set))
+  (define in-out-flow (create-data-set))
+  (mrs:map
+   (lambda (key 2d-vec)
+     (let ((x (car 2d-vec)) (y (cadr 2d-vec)))
+       (mrs:emit key (sqrt (+ (* x x) (* y y))))))
+   2d-vectors
+   magnitudes)
+  (mrs:map
+   (lambda (key 3d-vec)
+     (let ((x (car 3d-vec)) (y (cadr 3d-vec)) (z (caddr 3d-vec)))
+       (mrs:emit key (sqrt (+ (* x x) (* y y) (* z z))))))
+   3d-vectors
+   magnitudes)
+  (mrs:reduce
+   (lambda (key values)
+     (mrs:emit key (apply + values)))
+   magnitudes
+   in-out-flow)
+  (mrs:feed-key-value-list 2d-vectors '((in (1 2))
+                                        (in (3 3))
+                                        (out (5 0))))
+  (mrs:feed-key-value-list 3d-vectors '((in (1 1 1))
+                                        (out (5 0 0))))
+  (mrs:print-streaming in-out-flow 'flows))
+(mrs:run-computation demo)

examples/word-count.scm

@@ -0,0 +1,17 @@
+(define (word-count)
+  (define documents (create-data-set))
+  (define words (create-data-set))
+  (define word-counts (create-data-set))
+  (mrs:map
+   (lambda (key document)
+     (for-each (lambda (word) (mrs:emit word 1)) document))
+   documents words)
+  (mrs:reduce
+   (lambda (word counts)
+     (mrs:emit word (apply + counts)))
+   words word-counts)
+  (mrs:feed-value-list documents '((hello world this is a document)))
+  (mrs:feed-value-list documents '((another document) (and another)))
+  (mrs:feed-value-list documents '((this document is somewhat long) (a short document)))
+  (mrs:print-streaming word-counts 'count))
+(mrs:run-computation word-count)

operations.scm

@@ -35,13 +35,9 @@
    (cons value (hash-table/get hash-table key '()))))
 
 (define (mrs:aggregate ds-in ds-out)
-  (pp 'mrs-aggregate)
   (define (mm-func emit)
-    (pp 'mm-func-aggregator)    
     (let ((data-from-mapper (make-equal-hash-table)))
-      (pp 'made-aggregator)
       (lambda (ds-elt)
-	(pp (list 'aggregate-data ds-elt))
         (if (not (ds-elt-done? ds-elt))
             (append-data-in-hashtable
 	     data-from-mapper
@@ -50,9 +46,7 @@
             (begin (hash-table/for-each
 		    data-from-mapper
 		    (lambda (key values)
-		      (pp (list 'ht-kv key values))
-		      (emit (create-ds-elt key values))
-		      ))
+		      (emit (create-ds-elt key values))))
 		   (emit ds-elt))))))
   (make-distributor mm-func ds-in ds-out 1))
 
@@ -69,4 +63,36 @@
       (if (ds-elt-done? ds-elt)
           (pp `(,tag done))
           (pp `(,tag ,(ds-elt-key ds-elt) ,(ds-elt-value ds-elt))))))
-  (make-distributor mm-func ds-in (create-sink-data-set) 1))
+  (make-distributor mm-func ds-in (create-sink-data-set) 1))
+
+
+;;; Constructor-style operations
+(define (make-constructor-operation operation)
+  (define (new-op . args)
+    (define ds-out (create-data-set))
+    (apply operation (append args (list ds-out)))
+    ds-out)
+  new-op)
+
+(define mrs:c-map (make-constructor-operation mrs:map))
+(define mrs:c-filter (make-constructor-operation mrs:filter))
+(define mrs:c-aggregate (make-constructor-operation mrs:aggregate))
+(define mrs:c-reduce (make-constructor-operation mrs:reduce))
+
+#|
+(define (test1)
+  (define ds-input (create-data-set))
+  (define ds-out
+    (mrs:c-map
+     (lambda (key value)
+       (mrs:emit key (* 10 value)))
+     ds-input))
+  (mrs:print-streaming ds-out 'out)
+  (mrs:feed-value-list ds-input '(1 2 3 4)))
+(mrs:run-computation test1)
+;-> (out 0 10)
+;   (out 1 20)
+;   (out 2 30)
+;   (out 3 40)
+;   (out done)
+|#

workers.scm

@@ -8,7 +8,6 @@
         (input-pipe (make-conspire-pipe)))
     (let ((output-writer (get-pipe-writer output-pipe))
           (input-reader (get-pipe-reader input-pipe)))
-      (pp 'make-worker)
       (conspire:make-thread
        conspire:runnable
        (lambda ()
@@ -59,7 +58,6 @@
         (ds-in-reader (ds-get-reader ds-in))
         (ds-out-writer (ds-get-writer ds-out))
 	(num-workers-outstanding 0))
-    (pp 'make-distributor)
     (conspire:make-thread
      conspire:runnable
      (lambda ()
@@ -69,16 +67,13 @@
            (cond ((empty-ds-elt? ds-in-elt)
                   'continue)
                  ((ds-elt-done? ds-in-elt)
-		  (pp (list 'ds-done-elt-into-distributor ds-in-elt))
 		  (set! num-workers-outstanding num-workers)
 		  (vector-for-each (lambda (worker)
 				     (write-to-worker
 				      worker
 				      ds-in-elt))
 				   workers))
                  (else
-		  (pp (list 'ds-elt-into-distributor-1 ds-in-elt))
-		  (pp (list 'ds-elt-into-distributor-2 (ds-elt-value ds-in-elt)))		  		  
                   (write-to-worker
                    (choose-worker ds-in-elt workers)
                    ds-in-elt))))