gistfile1.hs

-- in Stats.hs
processMessages :: Log -> [(S.ByteString, Int, Int)]
processMessages log = map fmt $ M.toList $ mapReduce rwhnf (foldl' update M.empty)
                                                     rwhnf (M.unionsWith (+)) ([msgs log])
    where
        -- Increments the user (word count discardedfor now) line count by 1.
        update map (Message _ nick _) = M.insertWith (+) nick 1 map
        msgs = filter isMessage
        -- FIXME: adds a word count of 1 to the (Nick, LineCount) pairs from M.toList
        -- add wordcount!
        fmt (a, b) = (a, 1, b)

-- MapReduce.hs
-- from Real World Haskell chapter 24
module Hisg.MapReduce
    (
      mapReduce
    , simpleMapReduce
    -- exported for convenience
    , rnf
    , rwhnf
    ) where

import Control.Parallel (pseq)
import Control.Parallel.Strategies

{-- snippet simpleMapReduce.type --}
simpleMapReduce
    :: (a -> b)      -- map function
    -> ([b] -> c)    -- reduce function
    -> [a]           -- list to map over
    -> c
{-- /snippet simpleMapReduce.type --}

{-- snippet simpleMapReduce --}
simpleMapReduce mapFunc reduceFunc = reduceFunc . map mapFunc
{-- /snippet simpleMapReduce --}

{-- snippet mapReduce.type --}
mapReduce
    :: Strategy b    -- evaluation strategy for mapping
    -> (a -> b)      -- map function
    -> Strategy c    -- evaluation strategy for reduction
    -> ([b] -> c)    -- reduce function
    -> [a]           -- list to map over
    -> c
{-- /snippet mapReduce.type --}

{-- snippet mapReduce --}
mapReduce mapStrat mapFunc reduceStrat reduceFunc input =
    mapResult `pseq` reduceResult
  where mapResult    = parMap mapStrat mapFunc input
        reduceResult = reduceFunc mapResult `using` reduceStrat
{-- /snippet mapReduce --}