git.iko.soy/distributed-process
on master help sign in
Partial implementation of K-Means
commit8e92096b25684b28f597b7913949081673c235dd authorEdsko de Vries <edsko@edsko.net> dateTue 25 Sep 2012, 17:13 (14 years ago) parent9374848
5 files changed+83−20
compare with master
unifiedsplit
distributed-process-demos/distributed-process-demos.cabal+3−1 distributed-process-demos/src/MapReduce/CountWords.hs+1−1 distributed-process-demos/src/MapReduce/KMeans.hs+40 distributed-process-demos/src/MapReduce/MapReduce.hs+17−17 distributed-process-demos/src/MapReduce/SimpleLocalnet.hs+22−1
distributed-process-demos/distributed-process-demos.cabal+3−1
@@ -70,7 +70,9 @@ executable distributed-process-simplelocalnet-mapreduce
7070 distributed-static >= 0.2 && < 0.3,
7171 containers >= 0.4 && < 0.6,
7272 bytestring >= 0.9 && < 0.11,
73− binary >= 0.5 && < 0.6
73+ binary >= 0.5 && < 0.6,
74+ array >= 0.4 && < 0.5,
75+ random >= 1.0 && < 1.1
7476 extensions: BangPatterns, CPP, TemplateHaskell, ScopedTypeVariables
7577 ghc-options: -Wall
7678 if flag(use-simplelocalnet)
distributed-process-demos/src/MapReduce/CountWords.hs+1−1
@@ -15,7 +15,7 @@ type Document = String
1515type Word = String
1616type Frequency = Int
1717
18−countWords :: MapReduce FilePath Document Word Frequency
18+countWords :: MapReduce FilePath Document Word Frequency Frequency
1919countWords = MapReduce {
2020 mrMap = const (map (, 1) . words)
2121 , mrReduce = const sum
distributed-process-demos/src/MapReduce/KMeans.hsnew file+40
@@ -0,0 +1,40 @@
1+module KMeans
2+ ( Point
3+ , localKMeans
4+ ) where
5+
6+import Data.List (minimumBy)
7+import Data.Function (on)
8+import Data.Array (Array, (!), bounds)
9+import qualified Data.Map as Map (fromList)
10+import MapReduce
11+
12+type Point = (Double, Double)
13+type Cluster = (Double, Double)
14+
15+average :: Fractional a => [a] -> a
16+average xs = sum xs / fromIntegral (length xs)
17+
18+distanceSq :: Point -> Point -> Double
19+distanceSq (x1, y1) (x2, y2) = a * a + b * b
20+ where
21+ a = x2 - x1
22+ b = y2 - y1
23+
24+nearest :: Point -> [Cluster] -> Cluster
25+nearest p = minimumBy (compare `on` distanceSq p)
26+
27+center :: [Point] -> Point
28+center ps = let (xs, ys) = unzip ps in (average xs, average ys)
29+
30+kmeans :: Array Int Point -> MapReduce (Int, Int) [Cluster] Cluster Point Point
31+kmeans points = MapReduce {
32+ mrMap = \(lo, hi) cs -> [ let p = points ! i in (nearest p cs, p)
33+ | i <- [lo .. hi]
34+ ]
35+ , mrReduce = \_ ps -> center ps
36+ }
37+
38+localKMeans :: Array Int Point -> [Cluster] -> Map Cluster Point
39+localKMeans points cs =
40+ localMapReduce (kmeans points) (Map.fromList [(bounds points, cs)])
distributed-process-demos/src/MapReduce/MapReduce.hs+17−17
@@ -28,9 +28,9 @@ import Control.Concurrent (threadDelay)
2828--------------------------------------------------------------------------------
2929
3030-- | MapReduce skeleton
31−data MapReduce k1 v1 k2 v2 = MapReduce {
31+data MapReduce k1 v1 k2 v2 v3 = MapReduce {
3232 mrMap :: k1 -> v1 -> [(k2, v2)]
33− , mrReduce :: k2 -> [v2] -> v2
33+ , mrReduce :: k2 -> [v2] -> v3
3434 } deriving (Typeable)
3535
3636-- | Local (non-distributed) implementation of the map-reduce algorithm
@@ -38,19 +38,19 @@ data MapReduce k1 v1 k2 v2 = MapReduce {
3838-- This can be regarded as the specification of map-reduce; see
3939-- /Google’s MapReduce Programming Model---Revisited/ by Ralf Laemmel
4040-- (<http://userpages.uni-koblenz.de/~laemmel/MapReduce/>).
41−localMapReduce :: forall k1 k2 v1 v2. Ord k2 =>
42− MapReduce k1 v1 k2 v2
41+localMapReduce :: forall k1 k2 v1 v2 v3. Ord k2 =>
42+ MapReduce k1 v1 k2 v2 v3
4343 -> Map k1 v1
44− -> Map k2 v2
44+ -> Map k2 v3
4545localMapReduce mr = reducePerKey mr . groupByKey . mapPerKey mr
4646
47−reducePerKey :: MapReduce k1 v1 k2 v2 -> Map k2 [v2] -> Map k2 v2
47+reducePerKey :: MapReduce k1 v1 k2 v2 v3 -> Map k2 [v2] -> Map k2 v3
4848reducePerKey mr = Map.mapWithKey (mrReduce mr)
4949
5050groupByKey :: Ord k2 => [(k2, v2)] -> Map k2 [v2]
5151groupByKey = Map.fromListWith (++) . map (second return)
5252
53−mapPerKey :: MapReduce k1 v1 k2 v2 -> Map k1 v1 -> [(k2, v2)]
53+mapPerKey :: MapReduce k1 v1 k2 v2 v3 -> Map k1 v1 -> [(k2, v2)]
5454mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
5555
5656--------------------------------------------------------------------------------
@@ -63,11 +63,11 @@ matchDict SerializableDict = match
6363sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
6464sendDict SerializableDict = send
6565
66−mapperProcess :: forall k1 v1 k2 v2.
66+mapperProcess :: forall k1 v1 k2 v2 v3.
6767 SerializableDict (ProcessId, k1, v1)
6868 -> SerializableDict [(k2, v2)]
6969 -> ProcessId
70− -> MapReduce k1 v1 k2 v2
70+ -> MapReduce k1 v1 k2 v2 v3
7171 -> Process ()
7272mapperProcess dictIn dictOut workQueue mr = do
7373 us <- getSelfPid
@@ -97,30 +97,30 @@ mapperProcess dictIn dictOut workQueue mr = do
9797
9898remotable ['mapperProcess]
9999
100−mapperProcessClosure :: forall k1 v1 k2 v2.
101− (Typeable k1, Typeable v1, Typeable k2, Typeable v2)
100+mapperProcessClosure :: forall k1 v1 k2 v2 v3.
101+ (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
102102 => Static (SerializableDict (ProcessId, k1, v1))
103103 -> Static (SerializableDict [(k2, v2)])
104− -> Closure (MapReduce k1 v1 k2 v2)
104+ -> Closure (MapReduce k1 v1 k2 v2 v3)
105105 -> ProcessId
106106 -> Closure (Process ())
107107mapperProcessClosure dictIn dictOut mr pid =
108108 closure decoder (encode pid) `closureApply` mr
109109 where
110− decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 -> Process ())
110+ decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
111111 decoder =
112112 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
113113 `staticCompose`
114114 staticDecode sdictProcessId
115115
116−distrMapReduce :: forall k1 k2 v1 v2.
117− (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Ord k2)
116+distrMapReduce :: forall k1 k2 v1 v2 v3.
117+ (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
118118 => Static (SerializableDict (ProcessId, k1, v1))
119119 -> Static (SerializableDict [(k2, v2)])
120− -> Closure (MapReduce k1 v1 k2 v2)
120+ -> Closure (MapReduce k1 v1 k2 v2 v3)
121121 -> [NodeId]
122122 -> Map k1 v1
123− -> Process (Map k2 v2)
123+ -> Process (Map k2 v3)
124124distrMapReduce dictIn dictOut mr mappers input = do
125125 us <- getSelfPid
126126 slavesTerminated <- liftIO $ newEmptyMVar
distributed-process-demos/src/MapReduce/SimpleLocalnet.hs+22−1
@@ -1,12 +1,17 @@
11import System.Environment (getArgs)
2+import Control.Applicative
3+import Control.Monad
4+import System.Random
25import Control.Distributed.Process
36import Control.Distributed.Process.Node (initRemoteTable)
47import Control.Distributed.Process.Backend.SimpleLocalnet
58import Data.Map (Map)
9+import Data.Array (Array, listArray)
610import qualified Data.Map as Map (fromList)
711
812import qualified CountWords
913import qualified MapReduce
14+import qualified KMeans
1015
1116rtable :: RemoteTable
1217rtable = MapReduce.__remoteTable
@@ -18,20 +23,36 @@ constructInput files = do
1823 contents <- mapM readFile files
1924 return . Map.fromList $ zip files contents
2025
26+randomPoint :: IO KMeans.Point
27+randomPoint = (,) <$> randomIO <*> randomIO
28+
29+arrayFromList :: [e] -> Array Int e
30+arrayFromList xs = listArray (0, length xs - 1) xs
31+
2132main :: IO ()
2233main = do
2334 args <- getArgs
2435
2536 case args of
37+ -- Local word count
2638 "local" : "count" : files -> do
2739 input <- constructInput files
28− print . CountWords.localCountWords $ input
40+ print $ CountWords.localCountWords input
41+
42+ -- Distributed word count
2943 "master" : "count" : host : port : files -> do
3044 input <- constructInput files
3145 backend <- initializeBackend host port rtable
3246 startMaster backend $ \slaves -> do
3347 result <- CountWords.distrCountWords slaves input
3448 liftIO $ print result
49+
50+ -- Local k-means
51+ "local" : "kmeans" : [] -> do
52+ points <- replicateM 1000 randomPoint
53+ print $ KMeans.localKMeans (arrayFromList points) (take 10 points)
54+
55+ -- Generic slave for distributed examples
3556 "slave" : host : port : [] -> do
3657 backend <- initializeBackend host port rtable
3758 startSlave backend