git.iko.soy/distributed-process
on master help sign in
Partial implementation of K-Means
8e92096Edsko de Vries
distributed-process-demos/distributed-process-demos.cabal+3−1
compare with master
unifiedsplit
@@ -70,7 +70,9 @@ executable distributed-process-simplelocalnet-mapreduce
70 distributed-static >= 0.2 && < 0.3,
70 distributed-static >= 0.2 && < 0.3,
71 containers >= 0.4 && < 0.6,
71 containers >= 0.4 && < 0.6,
72 bytestring >= 0.9 && < 0.11,
72 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
74 extensions: BangPatterns, CPP, TemplateHaskell, ScopedTypeVariables
76 extensions: BangPatterns, CPP, TemplateHaskell, ScopedTypeVariables
75 ghc-options: -Wall
77 ghc-options: -Wall
76 if flag(use-simplelocalnet)
78 if flag(use-simplelocalnet)
distributed-process-demos/src/MapReduce/CountWords.hs+1−1
@@ -15,7 +15,7 @@ type Document = String
15type Word = String
15type Word = String
16type Frequency = Int
16type Frequency = Int
17
17
18−countWords :: MapReduce FilePath Document Word Frequency
18+countWords :: MapReduce FilePath Document Word Frequency Frequency
19countWords = MapReduce {
19countWords = MapReduce {
20 mrMap = const (map (, 1) . words)
20 mrMap = const (map (, 1) . words)
21 , mrReduce = const sum
21 , 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)
28--------------------------------------------------------------------------------
28--------------------------------------------------------------------------------
29
29
30-- | MapReduce skeleton
30-- | MapReduce skeleton
31−data MapReduce k1 v1 k2 v2 = MapReduce {
31+data MapReduce k1 v1 k2 v2 v3 = MapReduce {
32 mrMap :: k1 -> v1 -> [(k2, v2)]
32 mrMap :: k1 -> v1 -> [(k2, v2)]
33− , mrReduce :: k2 -> [v2] -> v2
33+ , mrReduce :: k2 -> [v2] -> v3
34 } deriving (Typeable)
34 } deriving (Typeable)
35
35
36-- | Local (non-distributed) implementation of the map-reduce algorithm
36-- | Local (non-distributed) implementation of the map-reduce algorithm
@@ -38,19 +38,19 @@ data MapReduce k1 v1 k2 v2 = MapReduce {
38-- This can be regarded as the specification of map-reduce; see
38-- This can be regarded as the specification of map-reduce; see
39-- /Google’s MapReduce Programming Model---Revisited/ by Ralf Laemmel
39-- /Google’s MapReduce Programming Model---Revisited/ by Ralf Laemmel
40-- (<http://userpages.uni-koblenz.de/~laemmel/MapReduce/>).
40-- (<http://userpages.uni-koblenz.de/~laemmel/MapReduce/>).
41−localMapReduce :: forall k1 k2 v1 v2. Ord k2 =>
41+localMapReduce :: forall k1 k2 v1 v2 v3. Ord k2 =>
42− MapReduce k1 v1 k2 v2
42+ MapReduce k1 v1 k2 v2 v3
43 -> Map k1 v1
43 -> Map k1 v1
44− -> Map k2 v2
44+ -> Map k2 v3
45localMapReduce mr = reducePerKey mr . groupByKey . mapPerKey mr
45localMapReduce mr = reducePerKey mr . groupByKey . mapPerKey mr
46
46
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
48reducePerKey mr = Map.mapWithKey (mrReduce mr)
48reducePerKey mr = Map.mapWithKey (mrReduce mr)
49
49
50groupByKey :: Ord k2 => [(k2, v2)] -> Map k2 [v2]
50groupByKey :: Ord k2 => [(k2, v2)] -> Map k2 [v2]
51groupByKey = Map.fromListWith (++) . map (second return)
51groupByKey = Map.fromListWith (++) . map (second return)
52
52
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)]
54mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
54mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
55
55
56--------------------------------------------------------------------------------
56--------------------------------------------------------------------------------
@@ -63,11 +63,11 @@ matchDict SerializableDict = match
63sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
63sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
64sendDict SerializableDict = send
64sendDict SerializableDict = send
65
65
66−mapperProcess :: forall k1 v1 k2 v2.
66+mapperProcess :: forall k1 v1 k2 v2 v3.
67 SerializableDict (ProcessId, k1, v1)
67 SerializableDict (ProcessId, k1, v1)
68 -> SerializableDict [(k2, v2)]
68 -> SerializableDict [(k2, v2)]
69 -> ProcessId
69 -> ProcessId
70− -> MapReduce k1 v1 k2 v2
70+ -> MapReduce k1 v1 k2 v2 v3
71 -> Process ()
71 -> Process ()
72mapperProcess dictIn dictOut workQueue mr = do
72mapperProcess dictIn dictOut workQueue mr = do
73 us <- getSelfPid
73 us <- getSelfPid
@@ -97,30 +97,30 @@ mapperProcess dictIn dictOut workQueue mr = do
97
97
98remotable ['mapperProcess]
98remotable ['mapperProcess]
99
99
100−mapperProcessClosure :: forall k1 v1 k2 v2.
100+mapperProcessClosure :: forall k1 v1 k2 v2 v3.
101− (Typeable k1, Typeable v1, Typeable k2, Typeable v2)
101+ (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
102 => Static (SerializableDict (ProcessId, k1, v1))
102 => Static (SerializableDict (ProcessId, k1, v1))
103 -> Static (SerializableDict [(k2, v2)])
103 -> Static (SerializableDict [(k2, v2)])
104− -> Closure (MapReduce k1 v1 k2 v2)
104+ -> Closure (MapReduce k1 v1 k2 v2 v3)
105 -> ProcessId
105 -> ProcessId
106 -> Closure (Process ())
106 -> Closure (Process ())
107mapperProcessClosure dictIn dictOut mr pid =
107mapperProcessClosure dictIn dictOut mr pid =
108 closure decoder (encode pid) `closureApply` mr
108 closure decoder (encode pid) `closureApply` mr
109 where
109 where
110− decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 -> Process ())
110+ decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
111 decoder =
111 decoder =
112 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
112 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
113 `staticCompose`
113 `staticCompose`
114 staticDecode sdictProcessId
114 staticDecode sdictProcessId
115
115
116−distrMapReduce :: forall k1 k2 v1 v2.
116+distrMapReduce :: forall k1 k2 v1 v2 v3.
117− (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Ord k2)
117+ (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
118 => Static (SerializableDict (ProcessId, k1, v1))
118 => Static (SerializableDict (ProcessId, k1, v1))
119 -> Static (SerializableDict [(k2, v2)])
119 -> Static (SerializableDict [(k2, v2)])
120− -> Closure (MapReduce k1 v1 k2 v2)
120+ -> Closure (MapReduce k1 v1 k2 v2 v3)
121 -> [NodeId]
121 -> [NodeId]
122 -> Map k1 v1
122 -> Map k1 v1
123− -> Process (Map k2 v2)
123+ -> Process (Map k2 v3)
124distrMapReduce dictIn dictOut mr mappers input = do
124distrMapReduce dictIn dictOut mr mappers input = do
125 us <- getSelfPid
125 us <- getSelfPid
126 slavesTerminated <- liftIO $ newEmptyMVar
126 slavesTerminated <- liftIO $ newEmptyMVar
distributed-process-demos/src/MapReduce/SimpleLocalnet.hs+22−1
@@ -1,12 +1,17 @@
1import System.Environment (getArgs)
1import System.Environment (getArgs)
2+import Control.Applicative
3+import Control.Monad
4+import System.Random
2import Control.Distributed.Process
5import Control.Distributed.Process
3import Control.Distributed.Process.Node (initRemoteTable)
6import Control.Distributed.Process.Node (initRemoteTable)
4import Control.Distributed.Process.Backend.SimpleLocalnet
7import Control.Distributed.Process.Backend.SimpleLocalnet
5import Data.Map (Map)
8import Data.Map (Map)
9+import Data.Array (Array, listArray)
6import qualified Data.Map as Map (fromList)
10import qualified Data.Map as Map (fromList)
7
11
8import qualified CountWords
12import qualified CountWords
9import qualified MapReduce
13import qualified MapReduce
14+import qualified KMeans
10
15
11rtable :: RemoteTable
16rtable :: RemoteTable
12rtable = MapReduce.__remoteTable
17rtable = MapReduce.__remoteTable
@@ -18,20 +23,36 @@ constructInput files = do
18 contents <- mapM readFile files
23 contents <- mapM readFile files
19 return . Map.fromList $ zip files contents
24 return . Map.fromList $ zip files contents
20
25
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+
21main :: IO ()
32main :: IO ()
22main = do
33main = do
23 args <- getArgs
34 args <- getArgs
24
35
25 case args of
36 case args of
37+ -- Local word count
26 "local" : "count" : files -> do
38 "local" : "count" : files -> do
27 input <- constructInput files
39 input <- constructInput files
28− print . CountWords.localCountWords $ input
40+ print $ CountWords.localCountWords input
41+
42+ -- Distributed word count
29 "master" : "count" : host : port : files -> do
43 "master" : "count" : host : port : files -> do
30 input <- constructInput files
44 input <- constructInput files
31 backend <- initializeBackend host port rtable
45 backend <- initializeBackend host port rtable
32 startMaster backend $ \slaves -> do
46 startMaster backend $ \slaves -> do
33 result <- CountWords.distrCountWords slaves input
47 result <- CountWords.distrCountWords slaves input
34 liftIO $ print result
48 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
35 "slave" : host : port : [] -> do
56 "slave" : host : port : [] -> do
36 backend <- initializeBackend host port rtable
57 backend <- initializeBackend host port rtable
37 startSlave backend
58 startSlave backend