git.iko.soy/distributed-process
on master help sign in
Finish k-means example
6ad0728Edsko de Vries
distributed-process-demos/src/MapReduce/CountWords.hs+6−3
compare with master
unifiedsplit
@@ -33,6 +33,9 @@ dictOut = SerializableDict
33remotable ['dictIn, 'dictOut, 'countWords]
33remotable ['dictIn, 'dictOut, 'countWords]
34
34
35distrCountWords :: [NodeId] -> Map FilePath Document -> Process (Map Word Frequency)
35distrCountWords :: [NodeId] -> Map FilePath Document -> Process (Map Word Frequency)
36+distrCountWords mappers input =
36−distrCountWords = distrMapReduce $(mkStatic 'dictIn)
37+ distrMapReduce $(mkStatic 'dictIn)
37− $(mkStatic 'dictOut)
38+ $(mkStatic 'dictOut)
38− (staticClosure $(mkStatic 'countWords))
39+ (staticClosure $(mkStatic 'countWords))
40+ mappers
41+ (\iteration -> iteration input)
distributed-process-demos/src/MapReduce/KMeans.hs+90−5
@@ -1,14 +1,21 @@
1+{-# LANGUAGE TupleSections #-}
1module KMeans
2module KMeans
2 ( Point
3 ( Point
3 , Cluster
4 , Cluster
4 , localKMeans
5 , localKMeans
6+ , distrKMeans
7+ , createGnuPlot
8+ , __remoteTable
5 ) where
9 ) where
6
10
11+import System.IO
7import Data.List (minimumBy)
12import Data.List (minimumBy)
8import Data.Function (on)
13import Data.Function (on)
9import Data.Array (Array, (!), bounds)
14import Data.Array (Array, (!), bounds)
10−import qualified Data.Map as Map (fromList)
15+import qualified Data.Map as Map (fromList, elems, toList, size)
11−import MapReduce
16+import Control.Distributed.Process
17+import Control.Distributed.Process.Closure
18+import MapReduce hiding (__remoteTable)
12
19
13type Point = (Double, Double)
20type Point = (Double, Double)
14type Cluster = (Double, Double)
21type Cluster = (Double, Double)
@@ -36,6 +43,84 @@ kmeans points = MapReduce {
36 , mrReduce = \_ ps -> (ps, center ps)
43 , mrReduce = \_ ps -> (ps, center ps)
37 }
44 }
38
45
39−localKMeans :: Array Int Point -> [Cluster] -> Map Cluster ([Point], Point)
46+localKMeans :: Array Int Point
47+ -> [Cluster]
48+ -> Int
49+ -> Map Cluster ([Point], Point)
40−localKMeans points cs =
50+localKMeans points cs iterations = go (iterations - 1)
51+ where
52+ mr :: [Cluster] -> Map Cluster ([Point], Point)
41− localMapReduce (kmeans points) (Map.fromList [(bounds points, cs)])
53+ mr = localMapReduce (kmeans points) . trivialSegmentation
54+
55+ go :: Int -> Map Cluster ([Point], Point)
56+ go 0 = mr cs
57+ go n = mr . map snd . Map.elems . go $ n - 1
58+
59+ trivialSegmentation :: [Cluster] -> Map (Int, Int) [Cluster]
60+ trivialSegmentation cs' = Map.fromList [(bounds points, cs')]
61+
62+dictIn :: SerializableDict (ProcessId, (Int, Int), [Cluster])
63+dictIn = SerializableDict
64+
65+dictOut :: SerializableDict [(Cluster, Point)]
66+dictOut = SerializableDict
67+
68+remotable ['kmeans, 'dictIn, 'dictOut]
69+
70+distrKMeans :: Array Int Point
71+ -> [Cluster]
72+ -> [NodeId]
73+ -> Int
74+ -> Process (Map Cluster ([Point], Point))
75+distrKMeans points cs mappers iterations =
76+ distrMapReduce $(mkStatic 'dictIn)
77+ $(mkStatic 'dictOut)
78+ ($(mkClosure 'kmeans) points)
79+ mappers
80+ (go (iterations - 1))
81+ where
82+ go :: Int
83+ -> (Map (Int, Int) [Cluster] -> Process (Map Cluster ([Point], Point)))
84+ -> Process (Map Cluster ([Point], Point))
85+ go 0 iteration = do
86+ iteration (Map.fromList $ map (, cs) segments)
87+ go n iteration = do
88+ clusters <- go (n - 1) iteration
89+ let centers = map snd $ Map.elems clusters
90+ iteration (Map.fromList $ map (, centers) segments)
91+
92+ segments :: [(Int, Int)]
93+ segments = let (lo, _) = bounds points in dividePoints numPoints lo
94+
95+ dividePoints :: Int -> Int -> [(Int, Int)]
96+ dividePoints pointsLeft offset
97+ | pointsLeft <= pointsPerMapper = [(offset, offset + pointsLeft - 1)]
98+ | otherwise = let offset' = offset + pointsPerMapper in
99+ (offset, offset' - 1)
100+ : dividePoints (pointsLeft - pointsPerMapper) offset'
101+
102+ pointsPerMapper :: Int
103+ pointsPerMapper =
104+ ceiling (toRational numPoints / toRational (length mappers))
105+
106+ numPoints :: Int
107+ numPoints = let (lo, hi) = bounds points in hi - lo + 1
108+
109+-- | Create a gnuplot data file for the output of the k-means algorithm
110+--
111+-- To plot the data, use
112+--
113+-- > plot "<<filename>>" u 1:2:3 with points palette
114+createGnuPlot :: Map KMeans.Cluster ([KMeans.Point], KMeans.Point) -> Handle -> IO ()
115+createGnuPlot clusters h =
116+ mapM_ printPoint . flatten . zip colors . Map.toList $ clusters
117+ where
118+ printPoint (x, y, color) =
119+ hPutStrLn h $ show x ++ " " ++ show y ++ " " ++ show color
120+
121+ flatten :: [(Float, (KMeans.Cluster, ([KMeans.Point], KMeans.Point)))]
122+ -> [(Double, Double, Float)]
123+ flatten = concat . map (\(color, (_, (points, _))) -> map (\(x, y) -> (x, y, color)) points)
124+
125+ colors :: [Float]
126+ colors = [0, 1 / fromIntegral (Map.size clusters) .. 1]
distributed-process-demos/src/MapReduce/MapReduce.hs+41−26
@@ -17,6 +17,7 @@ import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
17import Control.Arrow (second)
17import Control.Arrow (second)
18import Control.Monad (forM_, forever, replicateM)
18import Control.Monad (forM_, forever, replicateM)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
20+import qualified Control.Concurrent.Chan as Chan
20import Control.Distributed.Process
21import Control.Distributed.Process
21import Control.Distributed.Process.Serializable (Serializable)
22import Control.Distributed.Process.Serializable (Serializable)
22import Control.Distributed.Process.Closure
23import Control.Distributed.Process.Closure
@@ -113,43 +114,57 @@ mapperProcessClosure dictIn dictOut mr pid =
113 `staticCompose`
114 `staticCompose`
114 staticDecode sdictProcessId
115 staticDecode sdictProcessId
115
116
116−distrMapReduce :: forall k1 k2 v1 v2 v3.
117+distrMapReduce :: forall k1 k2 v1 v2 v3 a.
117 (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
118 (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
118 => Static (SerializableDict (ProcessId, k1, v1))
119 => Static (SerializableDict (ProcessId, k1, v1))
119 -> Static (SerializableDict [(k2, v2)])
120 -> Static (SerializableDict [(k2, v2)])
120 -> Closure (MapReduce k1 v1 k2 v2 v3)
121 -> Closure (MapReduce k1 v1 k2 v2 v3)
121 -> [NodeId]
122 -> [NodeId]
122− -> Map k1 v1
123− -> Process (Map k2 v3)
123+ -> ((Map k1 v1 -> Process (Map k2 v3)) -> Process a)
124+ -> Process a
124−distrMapReduce dictIn dictOut mr mappers input = do
125+distrMapReduce dictIn dictOut mr mappers p = do
126+ mr' <- unClosure mr
125 us <- getSelfPid
127 us <- getSelfPid
128+ queue <- liftIO $ Chan.newChan
126 slavesTerminated <- liftIO $ newEmptyMVar
129 slavesTerminated <- liftIO $ newEmptyMVar
127−
130+
128 workQueue <- spawnLocal $ do
131 workQueue <- spawnLocal $ do
132+ let go :: Process ()
133+ go = do
134+ mWork <- liftIO $ Chan.readChan queue
135+ case mWork of
136+ Just (key, val) -> do
129− -- As long as there is work, return the next Fib to compute
137+ -- As long there is work, make it available to the mappers
130− forM_ (Map.toList input) $ \(key, val) -> do
131− them <- expect
138+ them <- expect
132− send them (us, key, val)
139+ send them (us, key, val)
133−
140+ go
134− -- After that, just report that the work is done
141+ Nothing -> do
142+ -- Tell the mappers to terminate
135− replicateM (length mappers) $ do
143+ replicateM (length mappers) $ do
136− pid <- expect
144+ them <- expect
137− send pid ()
145+ send them ()
138−
139− liftIO $ putMVar slavesTerminated ()
146+ liftIO $ putMVar slavesTerminated ()
147+ go
140
148
141 -- Start the mappers
149 -- Start the mappers
142− liftIO $ print mappers
143 forM_ mappers $ flip spawn (mapperProcessClosure dictIn dictOut mr workQueue)
150 forM_ mappers $ flip spawn (mapperProcessClosure dictIn dictOut mr workQueue)
144
151
145− liftIO $ threadDelay 1000000
152+ let iteration :: Map k1 v1 -> Process (Map k2 v3)
153+ iteration input = do
154+ -- Make work available to the mappers
155+ liftIO $ mapM_ (Chan.writeChan queue . Just) (Map.toList input)
146
156
147− -- Wait for the result from the mappers
157+ -- Wait for the partial results
148− partials <- replicateM (Map.size input) expect
158+ partials <- replicateM (Map.size input) expect
149−
150− -- Wait for the slaves to terminate
151− liftIO $ takeMVar slavesTerminated
152
159
153− -- We reduce on this node
160+ -- We reduce on this node
154− mr' <- unClosure mr
155− return (reducePerKey mr' . groupByKey . concat $ partials)
161+ return (reducePerKey mr' . groupByKey . concat $ partials)
162+
163+ result <- p iteration
164+
165+ -- Terminate the wrappers
166+ liftIO $ do
167+ Chan.writeChan queue Nothing
168+ takeMVar slavesTerminated
169+
170+ return result
distributed-process-demos/src/MapReduce/SimpleLocalnet.hs+11−28
@@ -8,7 +8,7 @@ import Control.Distributed.Process.Node (initRemoteTable)
8import Control.Distributed.Process.Backend.SimpleLocalnet
8import Control.Distributed.Process.Backend.SimpleLocalnet
9import Data.Map (Map)
9import Data.Map (Map)
10import Data.Array (Array, listArray)
10import Data.Array (Array, listArray)
11−import qualified Data.Map as Map (fromList, toList, size, elems)
11+import qualified Data.Map as Map (fromList)
12
12
13import qualified CountWords
13import qualified CountWords
14import qualified MapReduce
14import qualified MapReduce
@@ -17,6 +17,7 @@ import qualified KMeans
17rtable :: RemoteTable
17rtable :: RemoteTable
18rtable = MapReduce.__remoteTable
18rtable = MapReduce.__remoteTable
19 . CountWords.__remoteTable
19 . CountWords.__remoteTable
20+ . KMeans.__remoteTable
20 $ initRemoteTable
21 $ initRemoteTable
21
22
22main :: IO ()
23main :: IO ()
@@ -40,15 +41,16 @@ main = do
40 -- Local k-means
41 -- Local k-means
41 "local" : "kmeans" : [] -> do
42 "local" : "kmeans" : [] -> do
42 points <- replicateM 1000 randomPoint
43 points <- replicateM 1000 randomPoint
44+ withFile "plot.data" WriteMode $ KMeans.createGnuPlot $
43− let kmeans = KMeans.localKMeans (arrayFromList points)
45+ KMeans.localKMeans (arrayFromList points) (take 5 points) 5
44
46
45− let it 0 = kmeans (take 5 points)
47+ -- Distributed k-means
46− it n = let clusters = it (n - 1) in
48+ "master" : "kmeans" : host : port : [] -> do
47− kmeans (map snd $ Map.elems clusters)
49+ points <- replicateM 1000 randomPoint
48−
50+ backend <- initializeBackend host port rtable
49− forM_ ([0 .. 4] :: [Int]) $ \n ->
51+ startMaster backend $ \slaves -> do
52+ result <- KMeans.distrKMeans (arrayFromList points) (take 5 points) slaves 5
50− withFile ("plot" ++ show n) WriteMode $
53+ liftIO $ withFile "plot.data" WriteMode $ KMeans.createGnuPlot result
51− createGnuPlot (it n)
52
54
53 -- Generic slave for distributed examples
55 -- Generic slave for distributed examples
54 "slave" : host : port : [] -> do
56 "slave" : host : port : [] -> do
@@ -69,22 +71,3 @@ randomPoint = (,) <$> randomIO <*> randomIO
69
71
70arrayFromList :: [e] -> Array Int e
72arrayFromList :: [e] -> Array Int e
71arrayFromList xs = listArray (0, length xs - 1) xs
73arrayFromList xs = listArray (0, length xs - 1) xs
72−
73−-- | Create a gnuplot data file for the output of the k-means algorithm
74−--
75−-- To plot the data, use
76−--
77−-- > plot "<<filename>>" u 1:2:3 with points palette
78−createGnuPlot :: Map KMeans.Cluster ([KMeans.Point], KMeans.Point) -> Handle -> IO ()
79−createGnuPlot clusters h =
80− mapM_ printPoint . flatten . zip colors . Map.toList $ clusters
81− where
82− printPoint (x, y, color) =
83− hPutStrLn h $ show x ++ " " ++ show y ++ " " ++ show color
84−
85− flatten :: [(Float, (KMeans.Cluster, ([KMeans.Point], KMeans.Point)))]
86− -> [(Double, Double, Float)]
87− flatten = concat . map (\(color, (_, (points, _))) -> map (\(x, y) -> (x, y, color)) points)
88−
89− colors :: [Float]
90− colors = [0, 1 / fromIntegral (Map.size clusters) .. 1]