content: -- ----------------------------------------------------------------------------
  {- |
    Module     : Examples.MapReduce.WordFrequency.WordFrequency
    Copyright  : Copyright (C) 2008 Stefan Schmidt
    License    : MIT
  
    Maintainer : Stefan Schmidt (stefanschmidt@web.de)
    Stability  : experimental
    Portability: portable
    Version    : 0.1
  
  
  -}
  -- ----------------------------------------------------------------------------
  
  module Examples.MapReduce.WordFrequency.WordFrequency
  (
    wordFrequencyAction
  , wordFrequencyActionMap
  , wordFrequencyDemoJob
  , createWordFrequencyDemoFiles
  , textList
  )
  where
  
  import           Data.Binary
  --import           Holumbus.Common.MRBinary
  
  import           System.Log.Logger
  
  import qualified Holumbus.FileSystem.FileSystem as FS
  
  import qualified Holumbus.Data.AccuMap as AMap
  import qualified Holumbus.Data.KeyMap as KMap
  import           Holumbus.MapReduce.Types
  
  
  localLogger :: String
  localLogger = "Examples.MapReduce.WordFrequency.WordFrequency"
  
  
  -- ----------------------------------------------------------------------------
  -- Word-Frequency
  -- ----------------------------------------------------------------------------
  
  
  mapWordFrequency :: ActionEnvironment -> () -> String -> String -> IO [(String, Integer)]
  mapWordFrequency _ _ k v
    = do 
      infoM localLogger "mapWordFrequency"
      debugM localLogger $ show ("input: " ++ k ++ " - " ++ show v)
      let v' = map (\s -> (s,1)) $ words v
      debugM localLogger $ show $ "output: " ++ show v'
      return v'
  
  
  reduceWordFrequency :: ActionEnvironment -> () -> String -> [Integer] -> IO (Maybe Integer)
  reduceWordFrequency _ _ k vs 
    = do
      infoM localLogger "reduce/combine WordFrequency"
      debugM localLogger $ show ("input: " ++ k ++ " - " ++ show vs)
      let s = sum vs
      debugM localLogger $ show $ "output: " ++ show s
      return (Just s)
      
    
  partitionWordFrequency :: ActionEnvironment -> () -> Int -> [(String, Integer)] -> IO [(Int,[(String, Integer)])]
  partitionWordFrequency _ _ _ ls 
    = do
      infoM localLogger "partitionCountWords"
      debugM localLogger $ show ls
      -- calculate partition-Values
      let markedList = map (\t@(k,_) ->  (length k,t)) ls
      -- merge them
      let resultList = AMap.toList $ AMap.fromTupleList markedList
      return resultList
  
  
  -- ----------------------------------------------------------------------------
  -- Actions
  -- ----------------------------------------------------------------------------
  
  wordFrequencyAction
    :: ActionConfiguration 
         ()                                          -- state
         String String                               -- k1, v1
         String Integer                              -- k2, v2
         Integer                                     -- v3 == v2
         Integer                                     -- v4
  wordFrequencyAction
    = (defaultActionConfiguration "WORDFREQUENCY")
          { ac_Map     = Just mapAction
          , ac_Combine = Nothing
          , ac_Reduce  = Just reduceAction
          }
      where
        mapAction 
          = (defaultMapConfiguration mapWordFrequency)
              { mc_Partition = partitionWordFrequency }
        reduceAction
          = (defaultReduceConfiguration reduceWordFrequency)
              { rc_Partition = partitionWordFrequency }
  
  wordFrequencyActionMap :: ActionMap
  wordFrequencyActionMap
    = KMap.insert (readActionConfiguration wordFrequencyAction) $
      KMap.empty
  
  
  -- ----------------------------------------------------------------------------
  -- DemoJob
  -- ----------------------------------------------------------------------------
  
    
  wordFrequencyDemoJob :: JobInfo
  wordFrequencyDemoJob = 
    createJobInfoFromConfiguration
      wordFrequencyAction -- action config
      ()                  -- options
      textList            -- input (Tuples)
      []                  -- input (Files)
      1                   -- number of splitters
      2                   -- number of mappers
      1                   -- number of reducers
      1                   -- number of results
      TOTRawTuple         -- type of the result (file of raw)
      
      
  textList :: [(String,String)]
  textList
    = [("text1", "aaa bb c dd dd"),("text2", "aaa bb"),("text3", "aaa dd dd")]
  
  
  -- ----------------------------------------------------------------------------
  -- DemoFiles
  -- ----------------------------------------------------------------------------
  
  
  createWordFrequencyDemoFiles :: FS.FileSystem -> IO ()
  createWordFrequencyDemoFiles fs
    = do
      let c = encode "harddisk file"
      FS.createFile "file1.txt" c fs
  