@@ -1,14 +1,21 @@
1+{-# LANGUAGE TupleSections #-}
12module KMeans
23 ( Point
34 , Cluster
45 , localKMeans
6+ , distrKMeans
7+ , createGnuPlot
8+ , __remoteTable
59 ) where
610
11+import System.IO
712import Data.List (minimumBy)
813import Data.Function (on)
914import Data.Array (Array, (!), bounds)
10−import qualified Data.Map as Map (fromList)
11−import MapReduce
15+import qualified Data.Map as Map (fromList, elems, toList, size)
16+import Control.Distributed.Process
17+import Control.Distributed.Process.Closure
18+import MapReduce hiding (__remoteTable)
1219
1320type Point = (Double, Double)
1421type Cluster = (Double, Double)
@@ -36,6 +43,84 @@ kmeans points = MapReduce {
3643 , mrReduce = \_ ps -> (ps, center ps)
3744 }
3845
39−localKMeans :: Array Int Point -> [Cluster] -> Map Cluster ([Point], Point)
40−localKMeans points cs =
41− localMapReduce (kmeans points) (Map.fromList [(bounds points, cs)])
46+localKMeans :: Array Int Point
47+ -> [Cluster]
48+ -> Int
49+ -> Map Cluster ([Point], Point)
50+localKMeans points cs iterations = go (iterations - 1)
51+ where
52+ mr :: [Cluster] -> Map Cluster ([Point], Point)
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]